jketreno/backstory
1
0
Fork 0
Code Issues 12 Actions Releases Activity

Grafana, Prometheus, Loki, and metrics

Browse Source
This commit is contained in:
James Ketr 2025-05-06 17:23:21 -07:00
parent 02651412fe
commit 963cd7491e
29 changed files with 3082 additions and 379 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

0
cache/grafana/.keep vendored Normal file
View File

34
docker-compose.yml
View File

@ -162,9 +162,41 @@ services:
networks:
- internal
volumes:
# - ./prometheus.yml:/etc/prometheus/prometheus.yml
- ./prometheus.yml:/etc/prometheus/prometheus.yml
- ./cache/prometheus:/prometheus
grafana:
image: grafana/grafana-oss
container_name: grafana
restart: "always"
env_file:
- .env.grafana
ports:
- 3111:3000 # Grafana
networks:
- internal
volumes:
- ./prometheus.yml:/etc/prometheus/prometheus.yml
- ./cache/grafana:/var/lib/grafana
loki:
image: grafana/loki
container_name: loki
restart: "always"
# env_file:
# - .env.grafana
ports:
- 3211:3100 # Grafana
networks:
- internal
command:
- -config.file=/loki-config.yaml
volumes:
# - ./prometheus.yml:/etc/prometheus/prometheus.yml
- ./loki-config.yaml:/loki-config.yaml
- ./cache/loki:/loki
networks:
internal:

7
frontend/src/Message.tsx
View File

@ -92,7 +92,11 @@ const MessageMeta = (props: MessageMetaProps) => {
prompt_eval_count,
prompt_eval_duration,
} = props.metadata || {};
const message = props.messageProps.message;
const message: any = props.messageProps.message;
let llm_submission: string = "<|system|>\n"
llm_submission += message.system_prompt + "\n\n"
llm_submission += message.context_prompt
return (<>
{
@ -204,6 +208,7 @@ const MessageMeta = (props: MessageMetaProps) => {
</Box>
</AccordionSummary>
<AccordionDetails>
<Box sx={{ pb: 1 }}>Copy LLM submission: <CopyBubble content={llm_submission} /></Box>
<JsonView displayDataTypes={false} objectSortKeys={true} collapsed={1} value={message} style={{ fontSize: "0.8rem", maxHeight: "20rem", overflow: "auto" }}>
<JsonView.String
render={({ children, ...reset }) => {

50
src/resume-flow/architecture.md Normal file
View File

@ -0,0 +1,50 @@
# Backstory Resume Generation Architecture Overview
The system follows a carefully designed pipeline with isolated stages to prevent fabrication:
## System Architecture Overview
The system uses a pipeline of isolated analysis and generation steps:
1. **Stage 1: Isolated Analysis** (three sub-stages)
- **1A: Job Analysis** - Extracts requirements from job description only
- **1B: Candidate Analysis** - Catalogs qualifications from resume/context only
- **1C: Mapping Analysis** - Identifies legitimate matches between requirements and qualifications
2. **Stage 2: Resume Generation**
- Uses mapping output to create a tailored resume with evidence-based content
3. **Stage 3: Verification**
- Performs fact-checking to catch any remaining fabrications
## Stage 1: Isolated Analysis (three separate sub-stages)
1. **Job Analysis**: Extracts requirements from just the job description
2. **Candidate Analysis**: Catalogs qualifications from just the resume/context
3. **Mapping Analysis**: Identifies legitimate matches between requirements and qualifications
## Stage 2: Resume Generation
Creates a tailored resume using only verified information from the mapping
## Stage 3: Verification
1. Performs fact-checking to catch any remaining fabrications
2. Corrects issues if needed and re-verifies
### Key Anti-Fabrication Mechanisms
The system uses several techniques to prevent fabrication:
* **Isolation of Analysis Stages**: By analyzing the job and candidate separately, the system prevents the LLM from prematurely creating connections that might lead to fabrication.
* **Evidence Requirements**: Each qualification included must have explicit evidence from the original materials.
* **Conservative Transferability**: The system is instructed to be conservative when claiming skills are transferable.
* **Verification Layer**: A dedicated verification step acts as a safety check to catch any remaining fabrications.
* **Strict JSON Structures**: Using structured JSON formats ensures information flows properly between stages.
## Implementation Details
* **Prompt Engineering**: Each stage has carefully designed prompts with clear instructions and output formats.
* **Error Handling**: Comprehensive validation and error handling throughout the pipeline.
* **Correction Loop**: If verification fails, the system attempts to correct issues and re-verify.
* **Traceability**: Information in the final resume can be traced back to specific evidence in the original materials.

48
src/resume-flow/flow-mermaid.md Normal file
View File

@ -0,0 +1,48 @@
flowchart TD
subgraph "Stage 1: Isolated Analysis"
subgraph "Stage 1A: Job Analysis"
A1[Job Description Input] --> A2[Job Analysis LLM]
A2 --> A3[Job Requirements JSON]
end
subgraph "Stage 1B: Candidate Analysis"
B1[Resume & Context Input] --> B2[Candidate Analysis LLM]
B2 --> B3[Candidate Qualifications JSON]
end
subgraph "Stage 1C: Mapping Analysis"
C1[Job Requirements JSON] --> C2[Candidate Qualifications JSON]
C2 --> C3[Mapping Analysis LLM]
C3 --> C4[Skills Mapping JSON]
end
end
subgraph "Stage 2: Resume Generation"
D1[Skills Mapping JSON] --> D2[Original Resume Reference]
D2 --> D3[Resume Generation LLM]
D3 --> D4[Tailored Resume Draft]
end
subgraph "Stage 3: Verification"
E1[Skills Mapping JSON] --> E2[Original Materials]
E2 --> E3[Tailored Resume Draft]
E3 --> E4[Verification LLM]
E4 --> E5{Verification Check}
E5 -->|PASS| E6[Approved Resume]
E5 -->|FAIL| E7[Correction Instructions]
E7 --> D3
end
A3 --> C1
B3 --> C2
C4 --> D1
D4 --> E3
style A2 fill:#f9d77e,stroke:#333,stroke-width:2px
style B2 fill:#f9d77e,stroke:#333,stroke-width:2px
style C3 fill:#f9d77e,stroke:#333,stroke-width:2px
style D3 fill:#f9d77e,stroke:#333,stroke-width:2px
style E4 fill:#f9d77e,stroke:#333,stroke-width:2px
style E5 fill:#a3e4d7,stroke:#333,stroke-width:2px
style E6 fill:#aed6f1,stroke:#333,stroke-width:2px
style E7 fill:#f5b7b1,stroke:#333,stroke-width:2px

636
src/resume-flow/impl-full.md Normal file
View File

@ -0,0 +1,636 @@
from typing import Dict, List, Any, Optional, Union
import json
import logging
import re
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# Define LLM interface
def call_llm(prompt: str, temperature: float = 0.2) -> str:
"""
Call your LLM of choice with the given prompt and parameters.
Implement connection to your specific LLM provider here.
"""
# Replace with your actual LLM call implementation
pass
# Helper functions
def extract_json_from_text(text: str) -> str:
"""Extract JSON string from text that may contain other content."""
json_pattern = r'```json\s*([\s\S]*?)\s*```'
match = re.search(json_pattern, text)
if match:
return match.group(1).strip()
# Try to find JSON without the markdown code block
json_pattern = r'({[\s\S]*})'
match = re.search(json_pattern, text)
if match:
return match.group(1).strip()
raise ValueError("No JSON found in the response")
def validate_job_requirements(job_requirements: Dict) -> None:
"""Validate the structure of job requirements."""
required_keys = ["job_requirements"]
if not all(key in job_requirements for key in required_keys):
missing = [key for key in required_keys if key not in job_requirements]
raise ValueError(f"Missing required keys in job requirements: {missing}")
# Additional validation can be added here
def validate_candidate_qualifications(candidate_qualifications: Dict) -> None:
"""Validate the structure of candidate qualifications."""
required_keys = ["candidate_qualifications"]
if not all(key in candidate_qualifications for key in required_keys):
missing = [key for key in required_keys if key not in candidate_qualifications]
raise ValueError(f"Missing required keys in candidate qualifications: {missing}")
# Additional validation can be added here
def validate_skills_mapping(skills_mapping: Dict) -> None:
"""Validate the structure of skills mapping."""
required_keys = ["skills_mapping", "resume_recommendations"]
if not all(key in skills_mapping for key in required_keys):
missing = [key for key in required_keys if key not in skills_mapping]
raise ValueError(f"Missing required keys in skills mapping: {missing}")
# Additional validation can be added here
def extract_header_from_resume(resume: str) -> str:
"""Extract header information from the original resume."""
# Simple implementation - in practice, you might want a more sophisticated approach
lines = resume.strip().split("\n")
# Take the first few non-empty lines as the header
header_lines = []
for line in lines[:10]: # Arbitrarily choose first 10 lines to search
if line.strip():
header_lines.append(line)
if len(header_lines) >= 4: # Assume header is no more than 4 lines
break
return "\n".join(header_lines)
# Stage 1A: Job Analysis Implementation
def create_job_analysis_prompt(job_description: str) -> str:
"""Create the prompt for job requirements analysis."""
system_prompt = """
You are an objective job requirements analyzer. Your task is to extract and categorize the specific skills,
experiences, and qualifications required in a job description WITHOUT any reference to any candidate.
## INSTRUCTIONS:
1. Analyze ONLY the job description provided.
2. Extract and categorize all requirements and preferences.
3. DO NOT consider any candidate information - this is a pure job analysis task.
## OUTPUT FORMAT:
```json
{
"job_requirements": {
"technical_skills": {
"required": ["skill1", "skill2"],
"preferred": ["skill1", "skill2"]
},
"experience_requirements": {
"required": ["exp1", "exp2"],
"preferred": ["exp1", "exp2"]
},
"education_requirements": ["req1", "req2"],
"soft_skills": ["skill1", "skill2"],
"industry_knowledge": ["knowledge1", "knowledge2"],
"responsibilities": ["resp1", "resp2"],
"company_values": ["value1", "value2"]
}
}
```
Be specific and detailed in your extraction. Break down compound requirements into individual components.
For example, "5+ years experience with React, Node.js and MongoDB" should be separated into:
- Experience: "5+ years software development"
- Technical skills: "React", "Node.js", "MongoDB"
Avoid vague categorizations and be precise about whether skills are explicitly required or just preferred.
"""
prompt = f"{system_prompt}\n\nJob Description:\n{job_description}"
return prompt
def analyze_job_requirements(job_description: str) -> Dict:
"""Analyze job requirements from job description."""
try:
prompt = create_job_analysis_prompt(job_description)
response = call_llm(prompt)
# Extract JSON from response
json_str = extract_json_from_text(response)
job_requirements = json.loads(json_str)
# Validate structure
validate_job_requirements(job_requirements)
return job_requirements
except Exception as e:
logger.error(f"Error in job requirements analysis: {str(e)}")
raise
# Stage 1B: Candidate Analysis Implementation
def create_candidate_analysis_prompt(resume: str, context: str) -> str:
"""Create the prompt for candidate qualifications analysis."""
system_prompt = """
You are an objective resume analyzer. Your task is to catalog ALL skills, experiences, and qualifications
present in a candidate's materials WITHOUT any reference to any job description.
## INSTRUCTIONS:
1. Analyze ONLY the candidate's resume and context provided.
2. Create a comprehensive inventory of the candidate's actual qualifications.
3. DO NOT consider any job requirements - this is a pure candidate analysis task.
4. For each qualification, cite exactly where in the materials it appears.
## OUTPUT FORMAT:
```json
{
"candidate_qualifications": {
"technical_skills": [
{
"skill": "skill name",
"evidence": "exact quote from materials",
"source": "resume or context",
"expertise_level": "explicit level mentioned or 'unspecified'"
}
],
"work_experience": [
{
"role": "job title",
"company": "company name",
"duration": "time period",
"responsibilities": ["resp1", "resp2"],
"technologies_used": ["tech1", "tech2"],
"achievements": ["achievement1", "achievement2"]
}
],
"education": [
{
"degree": "degree name",
"institution": "institution name",
"completed": true/false,
"evidence": "exact quote from materials"
}
],
"projects": [
{
"name": "project name",
"description": "brief description",
"technologies_used": ["tech1", "tech2"],
"evidence": "exact quote from materials"
}
],
"soft_skills": [
{
"skill": "skill name",
"evidence": "exact quote or inference basis",
"source": "resume or context"
}
]
}
}
```
Be thorough and precise. Include ONLY skills and experiences explicitly mentioned in the materials.
For each entry, provide the exact text evidence from the materials that supports its inclusion.
Do not make assumptions about skills based on job titles or project names - only include skills explicitly mentioned.
"""
prompt = f"{system_prompt}\n\nResume:\n{resume}\n\nAdditional Context:\n{context}"
return prompt
def analyze_candidate_qualifications(resume: str, context: str) -> Dict:
"""Analyze candidate qualifications from resume and context."""
try:
prompt = create_candidate_analysis_prompt(resume, context)
response = call_llm(prompt)
# Extract JSON from response
json_str = extract_json_from_text(response)
candidate_qualifications = json.loads(json_str)
# Validate structure
validate_candidate_qualifications(candidate_qualifications)
return candidate_qualifications
except Exception as e:
logger.error(f"Error in candidate qualifications analysis: {str(e)}")
raise
# Stage 1C: Mapping Analysis Implementation
def create_mapping_analysis_prompt(job_requirements: Dict, candidate_qualifications: Dict) -> str:
"""Create the prompt for mapping analysis."""
system_prompt = """
You are an objective skills mapper. Your task is to identify legitimate matches between job requirements
and candidate qualifications WITHOUT fabricating or stretching the truth.
## INSTRUCTIONS:
1. Use ONLY the structured job requirements and candidate qualifications provided.
2. Create a mapping that shows where the candidate's actual skills and experiences align with job requirements.
3. Identify gaps where the candidate lacks required qualifications.
4. Suggest legitimate transferable skills ONLY when there is reasonable evidence.
## OUTPUT FORMAT:
```json
{
"skills_mapping": {
"direct_matches": [
{
"job_requirement": "required skill",
"candidate_qualification": "matching skill",
"evidence": "exact quote from candidate materials"
}
],
"transferable_skills": [
{
"job_requirement": "required skill",
"candidate_qualification": "transferable skill",
"reasoning": "explanation of legitimate transferability",
"evidence": "exact quote from candidate materials"
}
],
"gap_analysis": {
"missing_required_skills": ["skill1", "skill2"],
"missing_preferred_skills": ["skill1", "skill2"],
"missing_experience": ["exp1", "exp2"]
}
},
"resume_recommendations": {
"highlight_points": [
{
"qualification": "candidate's qualification",
"relevance": "why this is highly relevant to the job"
}
],
"transferable_narratives": [
{
"from": "candidate's actual experience",
"to": "job requirement",
"suggested_framing": "how to honestly present the transfer"
}
],
"honest_limitations": [
"frank assessment of major qualification gaps"
]
}
}
```
CRITICAL RULES:
1. A "direct match" requires the EXACT SAME skill in both job requirements and candidate qualifications
2. A "transferable skill" must have legitimate, defensible connection - do not stretch credibility
3. All "missing_required_skills" MUST be acknowledged - do not ignore major gaps
4. Every match or transfer claim must cite specific evidence from the candidate materials
5. Be conservative in claiming transferability - when in doubt, list as missing
"""
prompt = f"{system_prompt}\n\nJob Requirements:\n{json.dumps(job_requirements, indent=2)}\n\n"
prompt += f"Candidate Qualifications:\n{json.dumps(candidate_qualifications, indent=2)}"
return prompt
def create_skills_mapping(job_requirements: Dict, candidate_qualifications: Dict) -> Dict:
"""Create mapping between job requirements and candidate qualifications."""
try:
prompt = create_mapping_analysis_prompt(job_requirements, candidate_qualifications)
response = call_llm(prompt)
# Extract JSON from response
json_str = extract_json_from_text(response)
skills_mapping = json.loads(json_str)
# Validate structure
validate_skills_mapping(skills_mapping)
return skills_mapping
except Exception as e:
logger.error(f"Error in skills mapping analysis: {str(e)}")
raise
# Stage 2: Resume Generation Implementation
def create_resume_generation_prompt(skills_mapping: Dict, candidate_qualifications: Dict, original_header: str) -> str:
"""Create the prompt for resume generation."""
system_prompt = """
You are a professional resume writer whose primary concern is FACTUAL ACCURACY. Your task is to create
a tailored resume that presents the candidate's actual qualifications in the most relevant way for this job,
using ONLY information that has been verified in the skills mapping.
## INSTRUCTIONS:
1. Use ONLY the information provided in the skills mapping JSON
2. Each skill, experience, or achievement you include MUST appear in either "direct_matches" or "transferable_skills"
3. DO NOT include skills listed in "missing_required_skills" or "missing_preferred_skills"
4. Format a professional resume with these sections:
- Header with name and contact information (exactly as provided in original resume)
- Professional Summary (focused on verified matching and transferable skills)
- Skills (ONLY from "direct_matches" and "transferable_skills" sections)
- Professional Experience (highlighting experiences referenced in the mapping)
- Education (exactly as listed in the candidate qualifications)
5. Follow these principles:
- Use the exact wording from "highlight_points" and "transferable_narratives" when describing experiences
- Maintain original job titles, companies, and dates exactly as provided
- Use achievement-oriented language that emphasizes results and impact
- Prioritize experiences that directly relate to the job requirements
## EVIDENCE REQUIREMENT:
For each skill or qualification you include in the resume, you MUST be able to trace it to:
1. A specific entry in "direct_matches" or "transferable_skills", AND
2. The original evidence citation in the candidate qualifications
If you cannot meet both these requirements for any content, DO NOT include it.
## FORMAT REQUIREMENTS:
- Create a clean, professional resume format
- Use consistent formatting for similar elements
- Ensure readability with appropriate white space
- Use bullet points for skills and achievements
- Include a final note: "Note: Initial draft of the resume was generated using the Backstory application written by James Ketrenos."
## FINAL VERIFICATION:
Before completing the resume:
1. Check that EVERY skill listed appears in either "direct_matches" or "transferable_skills"
2. Verify that no skills from "missing_required_skills" are included
3. Ensure all experience descriptions can be traced to evidence in candidate qualifications
4. Confirm that transferable skills are presented honestly without exaggeration
"""
prompt = f"{system_prompt}\n\nSkills Mapping:\n{json.dumps(skills_mapping, indent=2)}\n\n"
prompt += f"Candidate Qualifications:\n{json.dumps(candidate_qualifications, indent=2)}\n\n"
prompt += f"Original Resume Header:\n{original_header}"
return prompt
def generate_tailored_resume(skills_mapping: Dict, candidate_qualifications: Dict, original_header: str) -> str:
"""Generate a tailored resume based on skills mapping."""
try:
prompt = create_resume_generation_prompt(skills_mapping, candidate_qualifications, original_header)
response = call_llm(prompt, temperature=0.4) # Slightly higher temperature for better writing
return response
except Exception as e:
logger.error(f"Error in resume generation: {str(e)}")
raise
# Stage 3: Verification Implementation
def create_verification_prompt(generated_resume: str, skills_mapping: Dict, candidate_qualifications: Dict) -> str:
"""Create the prompt for resume verification."""
system_prompt = """
You are a critical resume fact-checker responsible for verifying the accuracy of a tailored resume.
Your task is to identify and flag any fabricated or embellished information that does not appear in
the candidate's original materials.
## INSTRUCTIONS:
1. Compare the tailored resume against:
- The structured skills mapping
- The candidate's original qualifications
2. Perform a line-by-line verification focusing on:
- Skills claimed vs. skills verified in original materials
- Experience descriptions vs. actual documented experience
- Projects and achievements vs. documented accomplishments
- Technical knowledge claims vs. verified technical background
3. Create a verification report with these sections:
## OUTPUT FORMAT:
```json
{
"verification_results": {
"factual_accuracy": {
"status": "PASS/FAIL",
"issues": [
{
"claim": "The specific claim in the resume",
"issue": "Why this is problematic",
"source_check": "Result of checking against source materials",
"suggested_correction": "How to fix this issue"
}
]
},
"skill_verification": {
"status": "PASS/FAIL",
"unverified_skills": ["skill1", "skill2"]
},
"experience_verification": {
"status": "PASS/FAIL",
"problematic_statements": [
{
"statement": "The problematic experience statement",
"issue": "Why this is problematic",
"suggested_correction": "How to fix this issue"
}
]
},
"overall_assessment": "APPROVED/NEEDS REVISION",
"correction_instructions": "Specific instructions for correcting the resume"
}
}
```
## CRITICAL VERIFICATION CRITERIA:
1. Any skill mentioned in the resume MUST appear verbatim in the skills mapping
2. Any technology experience claimed MUST be explicitly documented in original materials
3. Role descriptions must not imply expertise with technologies not listed in original materials
4. "Transferable skills" must be reasonably transferable, not stretches or fabrications
5. Job titles, dates, and companies must match exactly with original materials
6. Professional summary must not imply experience with technologies from the job description that aren't in the candidate's background
## SPECIAL ATTENTION:
Pay particular attention to subtle fabrications such as:
- Vague wording that implies experience ("worked with", "familiar with", "utilized") with technologies not in original materials
- Reframing unrelated experience to falsely imply relevance to the job requirements
- Adding technologies to project descriptions that weren't mentioned in the original materials
- Exaggerating level of involvement or responsibility in projects or roles
- Creating achievements that weren't documented in the original materials
"""
prompt = f"{system_prompt}\n\nTailored Resume:\n{generated_resume}\n\n"
prompt += f"Skills Mapping:\n{json.dumps(skills_mapping, indent=2)}\n\n"
prompt += f"Candidate Qualifications:\n{json.dumps(candidate_qualifications, indent=2)}"
return prompt
def verify_resume(generated_resume: str, skills_mapping: Dict, candidate_qualifications: Dict) -> Dict:
"""Verify the generated resume for accuracy against original materials."""
try:
prompt = create_verification_prompt(generated_resume, skills_mapping, candidate_qualifications)
response = call_llm(prompt)
# Extract JSON from response
json_str = extract_json_from_text(response)
verification_results = json.loads(json_str)
return verification_results
except Exception as e:
logger.error(f"Error in resume verification: {str(e)}")
raise
def correct_resume_issues(generated_resume: str, verification_results: Dict, skills_mapping: Dict, candidate_qualifications: Dict, original_header: str) -> str:
"""Correct issues in the resume based on verification results."""
if verification_results["verification_results"]["overall_assessment"] == "APPROVED":
return generated_resume
system_prompt = """
You are a professional resume editor with a focus on factual accuracy. Your task is to correct
the identified issues in a tailored resume according to the verification report.
## INSTRUCTIONS:
1. Make ONLY the changes specified in the verification report
2. Ensure all corrections maintain factual accuracy based on the skills mapping
3. Do not introduce any new claims or skills not present in the verification data
4. Maintain the original format and structure of the resume
## PROCESS:
1. For each issue in the verification report:
- Identify the problematic text in the resume
- Replace it with the suggested correction
- Ensure the correction is consistent with the rest of the resume
2. After making all corrections:
- Review the revised resume for consistency
- Ensure no factual inaccuracies have been introduced
- Check that all formatting remains professional
Return the fully corrected resume.
"""
prompt = f"{system_prompt}\n\nOriginal Resume:\n{generated_resume}\n\n"
prompt += f"Verification Results:\n{json.dumps(verification_results, indent=2)}\n\n"
prompt += f"Skills Mapping:\n{json.dumps(skills_mapping, indent=2)}\n\n"
prompt += f"Candidate Qualifications:\n{json.dumps(candidate_qualifications, indent=2)}\n\n"
prompt += f"Original Resume Header:\n{original_header}"
try:
response = call_llm(prompt, temperature=0.3)
return response
except Exception as e:
logger.error(f"Error in resume correction: {str(e)}")
raise
# Main orchestration function
def generate_factual_tailored_resume(job_description: str, resume: str, additional_context: str = "") -> Dict:
"""
Main function to generate a factually accurate tailored resume.
Args:
job_description: The job description text
resume: The candidate's original resume text
additional_context: Any additional context about the candidate (optional)
Returns:
Dict containing the generated resume and supporting analysis
"""
try:
logger.info("Starting multi-stage RAG resume generation process")
# Stage 1A: Analyze job requirements
logger.info("Stage 1A: Analyzing job requirements")
job_requirements = analyze_job_requirements(job_description)
# Stage 1B: Analyze candidate qualifications
logger.info("Stage 1B: Analyzing candidate qualifications")
candidate_qualifications = analyze_candidate_qualifications(resume, additional_context)
# Stage 1C: Create skills mapping
logger.info("Stage 1C: Creating skills mapping")
skills_mapping = create_skills_mapping(job_requirements, candidate_qualifications)
# Extract header from original resume
original_header = extract_header_from_resume(resume)
# Stage 2: Generate tailored resume
logger.info("Stage 2: Generating tailored resume")
generated_resume = generate_tailored_resume(skills_mapping, candidate_qualifications, original_header)
# Stage 3: Verify resume
logger.info("Stage 3: Verifying resume for accuracy")
verification_results = verify_resume(generated_resume, skills_mapping, candidate_qualifications)
# Handle corrections if needed
if verification_results["verification_results"]["overall_assessment"] == "NEEDS REVISION":
logger.info("Correcting issues found in verification")
generated_resume = correct_resume_issues(
generated_resume,
verification_results,
skills_mapping,
candidate_qualifications,
original_header
)
# Re-verify after corrections
logger.info("Re-verifying corrected resume")
verification_results = verify_resume(generated_resume, skills_mapping, candidate_qualifications)
# Return the final results
result = {
"job_requirements": job_requirements,
"candidate_qualifications": candidate_qualifications,
"skills_mapping": skills_mapping,
"generated_resume": generated_resume,
"verification_results": verification_results
}
logger.info("Resume generation process completed successfully")
return result
except Exception as e:
logger.error(f"Error in resume generation process: {str(e)}")
raise
# Command-line interface
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(description="Generate a factually accurate tailored resume")
parser.add_argument("--job", required=True, help="Path to job description file")
parser.add_argument("--resume", required=True, help="Path to resume file")
parser.add_argument("--context", help="Path to additional context file")
parser.add_argument("--output", default="output.json", help="Path to output file")
args = parser.parse_args()
# Read input files
with open(args.job, 'r') as f:
job_description = f.read()
with open(args.resume, 'r') as f:
resume = f.read()
additional_context = ""
if args.context:
with open(args.context, 'r') as f:
additional_context = f.read()
# Generate resume
result = generate_factual_tailored_resume(job_description, resume, additional_context)
# Write output
with open(args.output, 'w') as f:
json.dump(result, f, indent=2)
# Write the resume to a separate file for convenience
with open("generated_resume.txt", 'w') as f:
f.write(result["generated_resume"])
print(f"Resume generation complete. Results saved to {args.output}")
print(f"Generated resume saved to generated_resume.txt")

441
src/resume-flow/impl-part-one.md Normal file
View File

@ -0,0 +1,441 @@
# Complete Implementation Guide for Multi-Stage RAG Resume System
This guide provides a comprehensive implementation strategy for the multi-stage RAG resume generation system designed to prevent fabrication while creating relevant, tailored resumes.
## System Architecture Overview
The system uses a pipeline of isolated analysis and generation steps:
1. **Stage 1: Isolated Analysis** (three sub-stages)
- **1A: Job Analysis** - Extracts requirements from job description only
- **1B: Candidate Analysis** - Catalogs qualifications from resume/context only
- **1C: Mapping Analysis** - Identifies legitimate matches between requirements and qualifications
2. **Stage 2: Resume Generation**
- Uses mapping output to create a tailored resume with evidence-based content
3. **Stage 3: Verification**
- Performs fact-checking to catch any remaining fabrications
## Implementation Strategy
### 1. Setup and Prerequisites
```python
from typing import Dict, List, Any, Optional, Union
import json
import logging
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# Define LLM interface
def call_llm(prompt: str, temperature: float = 0.2) -> str:
"""
Call your LLM of choice with the given prompt and parameters.
Implement connection to your specific LLM provider here.
"""
# Replace with your actual LLM call implementation
pass
```
### 2. Stage 1A: Job Analysis Implementation
```python
def create_job_analysis_prompt(job_description: str) -> str:
"""Create the prompt for job requirements analysis."""
system_prompt = """
You are an objective job requirements analyzer. Your task is to extract and categorize the specific skills,
experiences, and qualifications required in a job description WITHOUT any reference to any candidate.
## INSTRUCTIONS:
1. Analyze ONLY the job description provided.
2. Extract and categorize all requirements and preferences.
3. DO NOT consider any candidate information - this is a pure job analysis task.
## OUTPUT FORMAT:
```json
{
"job_requirements": {
"technical_skills": {
"required": ["skill1", "skill2"],
"preferred": ["skill1", "skill2"]
},
"experience_requirements": {
"required": ["exp1", "exp2"],
"preferred": ["exp1", "exp2"]
},
"education_requirements": ["req1", "req2"],
"soft_skills": ["skill1", "skill2"],
"industry_knowledge": ["knowledge1", "knowledge2"],
"responsibilities": ["resp1", "resp2"],
"company_values": ["value1", "value2"]
}
}
```
Be specific and detailed in your extraction. Break down compound requirements into individual components.
For example, "5+ years experience with React, Node.js and MongoDB" should be separated into:
- Experience: "5+ years software development"
- Technical skills: "React", "Node.js", "MongoDB"
Avoid vague categorizations and be precise about whether skills are explicitly required or just preferred.
"""
prompt = f"{system_prompt}\n\nJob Description:\n{job_description}"
return prompt
def analyze_job_requirements(job_description: str) -> Dict:
"""Analyze job requirements from job description."""
try:
prompt = create_job_analysis_prompt(job_description)
response = call_llm(prompt)
# Extract JSON from response
json_str = extract_json_from_text(response)
job_requirements = json.loads(json_str)
# Validate structure
validate_job_requirements(job_requirements)
return job_requirements
except Exception as e:
logger.error(f"Error in job requirements analysis: {str(e)}")
raise
```
### 3. Stage 1B: Candidate Analysis Implementation
```python
def create_candidate_analysis_prompt(resume: str, context: str) -> str:
"""Create the prompt for candidate qualifications analysis."""
system_prompt = """
You are an objective resume analyzer. Your task is to catalog ALL skills, experiences, and qualifications
present in a candidate's materials WITHOUT any reference to any job description.
## INSTRUCTIONS:
1. Analyze ONLY the candidate's resume and context provided.
2. Create a comprehensive inventory of the candidate's actual qualifications.
3. DO NOT consider any job requirements - this is a pure candidate analysis task.
4. For each qualification, cite exactly where in the materials it appears.
## OUTPUT FORMAT:
```json
{
"candidate_qualifications": {
"technical_skills": [
{
"skill": "skill name",
"evidence": "exact quote from materials",
"source": "resume or context",
"expertise_level": "explicit level mentioned or 'unspecified'"
}
],
"work_experience": [
{
"role": "job title",
"company": "company name",
"duration": "time period",
"responsibilities": ["resp1", "resp2"],
"technologies_used": ["tech1", "tech2"],
"achievements": ["achievement1", "achievement2"]
}
],
"education": [
{
"degree": "degree name",
"institution": "institution name",
"completed": true/false,
"evidence": "exact quote from materials"
}
],
"projects": [
{
"name": "project name",
"description": "brief description",
"technologies_used": ["tech1", "tech2"],
"evidence": "exact quote from materials"
}
],
"soft_skills": [
{
"skill": "skill name",
"evidence": "exact quote or inference basis",
"source": "resume or context"
}
]
}
}
```
Be thorough and precise. Include ONLY skills and experiences explicitly mentioned in the materials.
For each entry, provide the exact text evidence from the materials that supports its inclusion.
Do not make assumptions about skills based on job titles or project names - only include skills explicitly mentioned.
"""
prompt = f"{system_prompt}\n\nResume:\n{resume}\n\nAdditional Context:\n{context}"
return prompt
def analyze_candidate_qualifications(resume: str, context: str) -> Dict:
"""Analyze candidate qualifications from resume and context."""
try:
prompt = create_candidate_analysis_prompt(resume, context)
response = call_llm(prompt)
# Extract JSON from response
json_str = extract_json_from_text(response)
candidate_qualifications = json.loads(json_str)
# Validate structure
validate_candidate_qualifications(candidate_qualifications)
return candidate_qualifications
except Exception as e:
logger.error(f"Error in candidate qualifications analysis: {str(e)}")
raise
```
### 4. Stage 1C: Mapping Analysis Implementation
```python
def create_mapping_analysis_prompt(job_requirements: Dict, candidate_qualifications: Dict) -> str:
"""Create the prompt for mapping analysis."""
system_prompt = """
You are an objective skills mapper. Your task is to identify legitimate matches between job requirements
and candidate qualifications WITHOUT fabricating or stretching the truth.
## INSTRUCTIONS:
1. Use ONLY the structured job requirements and candidate qualifications provided.
2. Create a mapping that shows where the candidate's actual skills and experiences align with job requirements.
3. Identify gaps where the candidate lacks required qualifications.
4. Suggest legitimate transferable skills ONLY when there is reasonable evidence.
## OUTPUT FORMAT:
```json
{
"skills_mapping": {
"direct_matches": [
{
"job_requirement": "required skill",
"candidate_qualification": "matching skill",
"evidence": "exact quote from candidate materials"
}
],
"transferable_skills": [
{
"job_requirement": "required skill",
"candidate_qualification": "transferable skill",
"reasoning": "explanation of legitimate transferability",
"evidence": "exact quote from candidate materials"
}
],
"gap_analysis": {
"missing_required_skills": ["skill1", "skill2"],
"missing_preferred_skills": ["skill1", "skill2"],
"missing_experience": ["exp1", "exp2"]
}
},
"resume_recommendations": {
"highlight_points": [
{
"qualification": "candidate's qualification",
"relevance": "why this is highly relevant to the job"
}
],
"transferable_narratives": [
{
"from": "candidate's actual experience",
"to": "job requirement",
"suggested_framing": "how to honestly present the transfer"
}
],
"honest_limitations": [
"frank assessment of major qualification gaps"
]
}
}
```
CRITICAL RULES:
1. A "direct match" requires the EXACT SAME skill in both job requirements and candidate qualifications
2. A "transferable skill" must have legitimate, defensible connection - do not stretch credibility
3. All "missing_required_skills" MUST be acknowledged - do not ignore major gaps
4. Every match or transfer claim must cite specific evidence from the candidate materials
5. Be conservative in claiming transferability - when in doubt, list as missing
"""
prompt = f"{system_prompt}\n\nJob Requirements:\n{json.dumps(job_requirements, indent=2)}\n\n"
prompt += f"Candidate Qualifications:\n{json.dumps(candidate_qualifications, indent=2)}"
return prompt
def create_skills_mapping(job_requirements: Dict, candidate_qualifications: Dict) -> Dict:
"""Create mapping between job requirements and candidate qualifications."""
try:
prompt = create_mapping_analysis_prompt(job_requirements, candidate_qualifications)
response = call_llm(prompt)
# Extract JSON from response
json_str = extract_json_from_text(response)
skills_mapping = json.loads(json_str)
# Validate structure
validate_skills_mapping(skills_mapping)
return skills_mapping
except Exception as e:
logger.error(f"Error in skills mapping analysis: {str(e)}")
raise
```
### 5. Stage 2: Resume Generation Implementation
```python
def create_resume_generation_prompt(skills_mapping: Dict, candidate_qualifications: Dict, original_header: str) -> str:
"""Create the prompt for resume generation."""
system_prompt = """
You are a professional resume writer whose primary concern is FACTUAL ACCURACY. Your task is to create
a tailored resume that presents the candidate's actual qualifications in the most relevant way for this job,
using ONLY information that has been verified in the skills mapping.
## INSTRUCTIONS:
1. Use ONLY the information provided in the skills mapping JSON
2. Each skill, experience, or achievement you include MUST appear in either "direct_matches" or "transferable_skills"
3. DO NOT include skills listed in "missing_required_skills" or "missing_preferred_skills"
4. Format a professional resume with these sections:
- Header with name and contact information (exactly as provided in original resume)
- Professional Summary (focused on verified matching and transferable skills)
- Skills (ONLY from "direct_matches" and "transferable_skills" sections)
- Professional Experience (highlighting experiences referenced in the mapping)
- Education (exactly as listed in the candidate qualifications)
5. Follow these principles:
- Use the exact wording from "highlight_points" and "transferable_narratives" when describing experiences
- Maintain original job titles, companies, and dates exactly as provided
- Use achievement-oriented language that emphasizes results and impact
- Prioritize experiences that directly relate to the job requirements
## EVIDENCE REQUIREMENT:
For each skill or qualification you include in the resume, you MUST be able to trace it to:
1. A specific entry in "direct_matches" or "transferable_skills", AND
2. The original evidence citation in the candidate qualifications
If you cannot meet both these requirements for any content, DO NOT include it.
## FORMAT REQUIREMENTS:
- Create a clean, professional resume format
- Use consistent formatting for similar elements
- Ensure readability with appropriate white space
- Use bullet points for skills and achievements
- Include a final note: "Note: Initial draft of the resume was generated using the Backstory application written by James Ketrenos."
## FINAL VERIFICATION:
Before completing the resume:
1. Check that EVERY skill listed appears in either "direct_matches" or "transferable_skills"
2. Verify that no skills from "missing_required_skills" are included
3. Ensure all experience descriptions can be traced to evidence in candidate qualifications
4. Confirm that transferable skills are presented honestly without exaggeration
"""
prompt = f"{system_prompt}\n\nSkills Mapping:\n{json.dumps(skills_mapping, indent=2)}\n\n"
prompt += f"Candidate Qualifications:\n{json.dumps(candidate_qualifications, indent=2)}\n\n"
prompt += f"Original Resume Header:\n{original_header}"
return prompt
def generate_tailored_resume(skills_mapping: Dict, candidate_qualifications: Dict, original_header: str) -> str:
"""Generate a tailored resume based on skills mapping."""
try:
prompt = create_resume_generation_prompt(skills_mapping, candidate_qualifications, original_header)
response = call_llm(prompt, temperature=0.4) # Slightly higher temperature for better writing
return response
except Exception as e:
logger.error(f"Error in resume generation: {str(e)}")
raise
```
### 6. Stage 3: Verification Implementation
```python
def create_verification_prompt(skills_mapping: Dict, original_materials: Dict, tailored_resume: str) -> str:
"""Create the prompt for resume verification."""
system_prompt = """
You are a critical resume fact-checker responsible for verifying the accuracy of a tailored resume.
Your task is to identify and flag any fabricated or embellished information that does not appear in
the candidate's original materials.
## INSTRUCTIONS:
1. Compare the tailored resume against:
- The structured skills mapping
- The candidate's original qualifications
2. Perform a line-by-line verification focusing on:
- Skills claimed vs. skills verified in original materials
- Experience descriptions vs. actual documented experience
- Projects and achievements vs. documented accomplishments
- Technical knowledge claims vs. verified technical background
3. Create a verification report with these sections:
## OUTPUT FORMAT:
```json
{
"verification_results": {
"factual_accuracy": {
"status": "PASS/FAIL",
"issues": [
{
"claim": "The specific claim in the resume",
"issue": "Why this is problematic",
"source_check": "Result of checking against source materials",
"suggested_correction": "How to fix this issue"
}
]
},
"skill_verification": {
"status": "PASS/FAIL",
"unverified_skills": ["skill1", "skill2"]
},
"experience_verification": {
"status": "PASS/FAIL",
"problematic_statements": [
{
"statement": "The problematic experience statement",
"issue": "Why this is problematic",
"suggested_correction": "How to fix this issue"
}
]
},
"overall_assessment": "APPROVED/NEEDS REVISION",
"correction_instructions": "Specific instructions for correcting the resume"
}
}
```
## CRITICAL VERIFICATION CRITERIA:
1. Any skill mentioned in the resume MUST appear verbatim in the skills mapping
2. Any technology experience claimed MUST be explicitly documented in original materials
3. Role descriptions must not imply expertise with technologies not listed in original materials
4. "Transferable skills" must be reasonably transferable, not stretches or fabrications
5. Job titles, dates, and companies must match exactly with original materials
6. Professional summary must not imply experience with technologies from the job description that aren't in the candidate's background
## SPECIAL ATTENTION:
Pay particular attention to subtle fabrications such as:
- Vague wording that implies experience ("worked with", "familiar with", "utilized") with technologies not in original materials
- Reframing unrelated experience to falsely imply relevance to the job requirements
- Adding technologies to project descriptions that weren't mentioned in the original materials
-

225
src/resume-flow/stage-1.md Normal file
View File

@ -0,0 +1,225 @@
# Improved Stage 1: Isolated Analysis Approach
Instead of analyzing both job requirements and candidate qualifications in a single step, we'll split Stage 1 into two isolated sub-stages:
## Stage 1A: Job Requirements Analysis
````
<|system|>
You are an objective job requirements analyzer. Your task is to extract and categorize the specific skills, experiences, and qualifications required in a job description WITHOUT any reference to any candidate.
## INSTRUCTIONS:
1. Analyze ONLY the job description provided.
2. Extract and categorize all requirements and preferences.
3. DO NOT consider any candidate information - this is a pure job analysis task.
## OUTPUT FORMAT:
```json
{
"job_requirements": {
"technical_skills": {
"required": ["skill1", "skill2"...],
"preferred": ["skill1", "skill2"...]
},
"experience_requirements": {
"required": ["exp1", "exp2"...],
"preferred": ["exp1", "exp2"...]
},
"education_requirements": ["req1", "req2"...],
"soft_skills": ["skill1", "skill2"...],
"industry_knowledge": ["knowledge1", "knowledge2"...],
"responsibilities": ["resp1", "resp2"...],
"company_values": ["value1", "value2"...]
}
}
```
Be specific and detailed in your extraction. Break down compound requirements into individual components.
For example, "5+ years experience with React, Node.js and MongoDB" should be separated into:
- Experience: "5+ years software development"
- Technical skills: "React", "Node.js", "MongoDB"
Avoid vague categorizations and be precise about whether skills are explicitly required or just preferred.
</|system|>
<|job_description|>
[INSERT JOB DESCRIPTION HERE]
</|job_description|>
````
## Stage 1B: Candidate Qualifications Inventory
````
<|system|>
You are an objective resume analyzer. Your task is to catalog ALL skills, experiences, and qualifications present in a candidate's materials WITHOUT any reference to any job description.
## INSTRUCTIONS:
1. Analyze ONLY the candidate's resume and context provided.
2. Create a comprehensive inventory of the candidate's actual qualifications.
3. DO NOT consider any job requirements - this is a pure candidate analysis task.
4. For each qualification, cite exactly where in the materials it appears.
## OUTPUT FORMAT:
```json
{
"candidate_qualifications": {
"technical_skills": [
{
"skill": "skill name",
"evidence": "exact quote from materials",
"source": "resume or context",
"expertise_level": "explicit level mentioned or 'unspecified'"
}
],
"work_experience": [
{
"role": "job title",
"company": "company name",
"duration": "time period",
"responsibilities": ["resp1", "resp2"...],
"technologies_used": ["tech1", "tech2"...],
"achievements": ["achievement1", "achievement2"...]
}
],
"education": [
{
"degree": "degree name",
"institution": "institution name",
"completed": true/false,
"evidence": "exact quote from materials"
}
],
"projects": [
{
"name": "project name",
"description": "brief description",
"technologies_used": ["tech1", "tech2"...],
"evidence": "exact quote from materials"
}
],
"soft_skills": [
{
"skill": "skill name",
"evidence": "exact quote or inference basis",
"source": "resume or context"
}
]
}
}
```
Be thorough and precise. Include ONLY skills and experiences explicitly mentioned in the materials.
For each entry, provide the exact text evidence from the materials that supports its inclusion.
Do not make assumptions about skills based on job titles or project names - only include skills explicitly mentioned.
</|system|>
<|resume|>
[INSERT RESUME HERE]
</|resume|>
<|context|>
[INSERT ADDITIONAL CONTEXT HERE]
</|context|>
````
## Stage 1C: Mapping Analysis
````
<|system|>
You are an objective skills mapper. Your task is to identify legitimate matches between job requirements and candidate qualifications WITHOUT fabricating or stretching the truth.
## INSTRUCTIONS:
1. Use ONLY the structured job requirements and candidate qualifications provided.
2. Create a mapping that shows where the candidate's actual skills and experiences align with job requirements.
3. Identify gaps where the candidate lacks required qualifications.
4. Suggest legitimate transferable skills ONLY when there is reasonable evidence.
## OUTPUT FORMAT:
```json
{
"skills_mapping": {
"direct_matches": [
{
"job_requirement": "required skill",
"candidate_qualification": "matching skill",
"evidence": "exact quote from candidate materials"
}
],
"transferable_skills": [
{
"job_requirement": "required skill",
"candidate_qualification": "transferable skill",
"reasoning": "explanation of legitimate transferability",
"evidence": "exact quote from candidate materials"
}
],
"gap_analysis": {
"missing_required_skills": ["skill1", "skill2"...],
"missing_preferred_skills": ["skill1", "skill2"...],
"missing_experience": ["exp1", "exp2"...]
}
},
"resume_recommendations": {
"highlight_points": [
{
"qualification": "candidate's qualification",
"relevance": "why this is highly relevant to the job"
}
],
"transferable_narratives": [
{
"from": "candidate's actual experience",
"to": "job requirement",
"suggested_framing": "how to honestly present the transfer"
}
],
"honest_limitations": [
"frank assessment of major qualification gaps"
]
}
}
```
CRITICAL RULES:
1. A "direct match" requires the EXACT SAME skill in both job requirements and candidate qualifications
2. A "transferable skill" must have legitimate, defensible connection - do not stretch credibility
3. All "missing_required_skills" MUST be acknowledged - do not ignore major gaps
4. Every match or transfer claim must cite specific evidence from the candidate materials
5. Be conservative in claiming transferability - when in doubt, list as missing
</|system|>
<|job_requirements|>
[INSERT OUTPUT FROM STAGE 1A HERE]
</|job_requirements|>
<|candidate_qualifications|>
[INSERT OUTPUT FROM STAGE 1B HERE]
</|candidate_qualifications|>
````
## Benefits of This Improved Approach:
1. **Complete Isolation**: By separating job analysis from candidate analysis, the model cannot mix requirements with qualifications
2. **Evidence-Based Qualifications**: Requiring specific citations for each qualification prevents fabrication
3. **Independent Gap Analysis**: With separate analyses, the gaps become explicit rather than being quietly filled with fabrications
4. **Traceable Reasoning**: Each claimed match or transferable skill must provide evidence and reasoning
5. **Clear Separation of Concerns**: Each sub-stage has a single focused task rather than attempting multiple analyses at once
## Implementation Flow:
1. Run Stage 1A to analyze job requirements in isolation
2. Run Stage 1B to catalog candidate qualifications in isolation
3. Run Stage 1C to map between requirements and qualifications
4. Proceed to Stage 2 with the mapping output
This approach leverages the "chain of thought" principle by breaking down the analysis into discrete steps with clear inputs and outputs between each step. The explicit evidence requirements and citation needs make it much harder for the model to fabricate connections.

81
src/resume-flow/stage-2.md Normal file
View File

@ -0,0 +1,81 @@
# Revised Stage 2: Evidence-Based Resume Generation
Now that we have isolated analyses and clear mapping between job requirements and candidate qualifications, we can create a more constrained resume generation prompt that works strictly from the verified mapping:
````
<|system|>
You are a professional resume writer whose primary concern is FACTUAL ACCURACY. Your task is to create a tailored resume that presents the candidate's actual qualifications in the most relevant way for this job, using ONLY information that has been verified in the skills mapping.
## INSTRUCTIONS:
1. Use ONLY the information provided in the skills mapping JSON
2. Each skill, experience, or achievement you include MUST appear in either "direct_matches" or "transferable_skills"
3. DO NOT include skills listed in "missing_required_skills" or "missing_preferred_skills"
4. Format a professional resume with these sections:
- Header with name and contact information (exactly as provided in original resume)
- Professional Summary (focused on verified matching and transferable skills)
- Skills (ONLY from "direct_matches" and "transferable_skills" sections)
- Professional Experience (highlighting experiences referenced in the mapping)
- Education (exactly as listed in the candidate qualifications)
5. Follow these principles:
- Use the exact wording from "highlight_points" and "transferable_narratives" when describing experiences
- Maintain original job titles, companies, and dates exactly as provided
- Use achievement-oriented language that emphasizes results and impact
- Prioritize experiences that directly relate to the job requirements
## EVIDENCE REQUIREMENT:
For each skill or qualification you include in the resume, you MUST be able to trace it to:
1. A specific entry in "direct_matches" or "transferable_skills", AND
2. The original evidence citation in the candidate qualifications
If you cannot meet both these requirements for any content, DO NOT include it.
## FORMAT REQUIREMENTS:
- Create a clean, professional resume format
- Use consistent formatting for similar elements
- Ensure readability with appropriate white space
- Use bullet points for skills and achievements
- Include the note at the end as specified in the original instructions
## FINAL VERIFICATION:
Before completing the resume:
1. Check that EVERY skill listed appears in either "direct_matches" or "transferable_skills"
2. Verify that no skills from "missing_required_skills" are included
3. Ensure all experience descriptions can be traced to evidence in candidate qualifications
4. Confirm that transferable skills are presented honestly without exaggeration
</|system|>
<|skills_mapping|>
[INSERT OUTPUT FROM STAGE 1C HERE]
</|skills_mapping|>
<|candidate_qualifications|>
[INSERT OUTPUT FROM STAGE 1B HERE]
</|candidate_qualifications|>
<|original_resume_header|>
[INSERT NAME AND CONTACT INFO FROM ORIGINAL RESUME]
</|original_resume_header|>
Please generate a professional, tailored resume that honestly represents the candidate's qualifications for this position based strictly on the verified skills mapping provided.
````
## Key Improvements in This Revised Stage 2:
1. **Evidence Traceability Requirement**: Every item in the resume must be traceable to both the skills mapping and the original evidence citation
2. **Explicit Allowlist**: The resume can only include skills explicitly listed in "direct_matches" or "transferable_skills"
3. **Explicit Blocklist**: The prompt specifically prohibits including skills from "missing_required_skills"
4. **Dual Verification**: Items must pass both the mapping check and have evidence in candidate qualifications
5. **Preservation of Original Details**: Job titles, companies, and dates must be kept exactly as they appeared in the original resume
6. **Final Verification Checklist**: The LLM must perform a final verification before finishing the resume
These constraints force the resume generation to work strictly from verified information while still allowing the LLM to present the candidate's qualifications in the most relevant and impactful way for the specific job.

288
src/resume-generation-flow.md Normal file
View File

@ -0,0 +1,288 @@
# Resume Generation Flow
![Resume Generation Flow](./resume-generation-flow.md)
## Stage 1: Job Description Analysis Prompt
````
<|system|>
You are an objective skills analyzer for resume tailoring. Your task is to analyze a job description and a candidate's background, identifying ONLY legitimate matches without fabrication. Your analysis will be used in a second stage to generate a tailored resume.
## INSTRUCTIONS:
1. Analyze the job description to extract:
- Required skills and technologies
- Required experience types
- Company values and culture indicators
- Primary responsibilities of the role
2. Analyze the candidate's resume and context to identify:
- Actual skills and technologies the candidate possesses
- Actual experience types the candidate has
- Relevant achievements that align with the job
3. Create a structured output with these sections:
### OUTPUT FORMAT:
```json
{
"job_requirements": {
"required_skills": ["skill1", "skill2"...],
"preferred_skills": ["skill1", "skill2"...],
"required_experience": ["exp1", "exp2"...],
"company_values": ["value1", "value2"...]
},
"candidate_qualifications": {
"matching_skills": ["skill1", "skill2"...],
"transferable_skills": ["skill1", "skill2"...],
"missing_skills": ["skill1", "skill2"...],
"relevant_experience": [
{
"role": "Role title",
"relevance": "Explanation of how this role is relevant"
}
],
"relevant_achievements": [
{
"achievement": "Achievement description",
"relevance": "Explanation of relevance to job"
}
]
},
"emphasis_recommendations": {
"highlight_skills": ["skill1", "skill2"...],
"highlight_experiences": ["exp1", "exp2"...],
"transferable_narratives": [
"Description of how candidate's experience in X transfers to required skill Y"
],
"honest_gap_acknowledgment": [
"Skills/technologies the candidate lacks with no reasonable transfer claim"
]
}
}
```
### CRITICAL RULES:
1. DO NOT INVENT OR IMPLY ANY SKILLS OR EXPERIENCE not explicitly present in the candidate's materials
2. A skill is only "matching" if the EXACT SAME technology/skill appears in both job requirements and candidate background
3. "Transferable skills" must have a legitimate connection - don't stretch credibility
4. Be brutally honest in "missing_skills" and "honest_gap_acknowledgment" sections
5. For each skill or experience match, note the specific section/line from the candidate's materials where it appears
6. If a job requirement has no legitimate match or transfer in the candidate's background, it MUST appear in "missing_skills"
The output of this analysis will be used to create a truthful, tailored resume. Accuracy is critical.
</|system|>
<|job_description|>
[INSERT JOB DESCRIPTION HERE]
</|job_description|>
<|resume|>
[INSERT RESUME HERE]
</|resume|>
<|context|>
[INSERT ADDITIONAL CONTEXT HERE]
</|context|>
````
### How This Helps:
This first-stage prompt:
1. **Creates a clear separation** between analysis and resume generation
2. **Explicitly categorizes** what's matching, what's transferable, and what's missing
3. **Requires documentation** of where each skill/experience appears in the candidate materials
4. **Forces honesty** about skill gaps instead of fabricating experience
5. **Produces structured data** that can be used as controlled input for the resume generation stage
The JSON output provides a clean, structured way to pass verified information to the second stage, making it harder for the LLM to fabricate information since it must work within the constraints of what was identified in this analysis phase.
## Stage 2: Resume Generation Prompt
```
<|system|>
You are a professional resume writer. Your task is to create a tailored resume based ONLY on the candidate's verified qualifications and the job analysis provided. Under no circumstances should you invent or embellish information.
## INSTRUCTIONS:
1. Use ONLY the structured job analysis provided in <|job_analysis|> to create the resume
2. Format a professional resume with the following sections:
- Header with name and contact information (from original resume)
- Professional Summary (focused on verified matching and transferable skills)
- Skills (ONLY from "matching_skills" and "transferable_skills" sections)
- Professional Experience (highlighting experiences from "relevant_experience" with emphasis on "highlight_experiences")
- Education (exactly as listed in original resume)
3. For each entry in the resume:
- ONLY include details verified in the analysis
- Emphasize aspects recommended in "emphasis_recommendations"
- DO NOT add any skills from "missing_skills" or "honest_gap_acknowledgment"
4. Use professional, concise language that highlights the candidate's strengths for this specific role based on their ACTUAL qualifications
## CRITICAL RULES:
1. DO NOT INVENT OR FABRICATE any experience, skills, or qualifications
2. DO NOT ADD ANY TECHNOLOGY, LANGUAGE, OR FRAMEWORK that isn't in "matching_skills" or "transferable_skills"
3. DO NOT CLAIM EXPERIENCE with technologies listed in "missing_skills"
4. If you're uncertain about including something, err on the side of exclusion
5. Focus on honest transferable skills and achievements rather than claiming direct experience with unfamiliar technologies
6. Maintain complete factual accuracy while presenting the candidate in the best light based on their ACTUAL qualifications
Before finalizing the resume, perform a verification check:
- Review each skill and experience to confirm it appears in the job analysis
- Remove any content that cannot be verified against the job analysis
- Ensure no technologies from "missing_skills" have been accidentally included
Remember: The goal is to create an HONEST, tailored resume that presents the candidate's actual qualifications in the most relevant way for this specific job.
</|system|>
<|job_analysis|>
[INSERT OUTPUT FROM STAGE 1 HERE]
</|job_analysis|>
<|original_resume|>
[INSERT ORIGINAL RESUME FOR REFERENCE]
</|original_resume|>
Please generate a professional, tailored resume that honestly represents the candidate's qualifications for this position.
```
### Implementation Notes:
This two-stage approach:
1. **Decouples analysis from generation** - By separating these steps, we create a verification checkpoint between identifying relevant experience and creating the resume
2. **Creates an explicit allowlist** - The resume generation can only use skills and experiences that were verified in the analysis stage
3. **Minimizes hallucination** - By providing structured data rather than full text from the first stage, we reduce the chance of the LLM pulling in unverified information
4. **Enables verification** - The resume generator has access to both the analysis and original resume, allowing for fact-checking
5. **Focuses on transferable skills** - Instead of fabricating experience with missing technologies, the prompt emphasizes legitimate transferable skills
## Integration Strategy:
To implement this in your RAG application:
1. Run the first prompt to get the structured job analysis JSON
2. Parse the JSON output to validate it has the expected structure
3. Pass the validated analysis to the second prompt along with the original resume
4. Optionally implement a third verification stage that compares the generated resume against both the analysis and original materials to catch any fabrications
This multi-stage approach with structured intermediary data will significantly reduce the tendency of the LLM to fabricate qualifications while still producing a relevant, tailored resume.
## Stage 3: Resume Verification Prompt
````
<|system|>
You are a critical resume fact-checker responsible for verifying the accuracy of a tailored resume. Your task is to identify and flag any fabricated or embellished information that does not appear in the candidate's original materials.
### INSTRUCTIONS:
1. Compare the tailored resume against:
- The structured job analysis
- The candidate's original resume and context
2. Perform a line-by-line verification focusing on:
- Skills claimed vs. skills verified in original materials
- Experience descriptions vs. actual documented experience
- Projects and achievements vs. documented accomplishments
- Technical knowledge claims vs. verified technical background
3. Create a verification report with these sections:
#### OUTPUT FORMAT:
```json
{
"verification_results": {
"factual_accuracy": {
"status": "PASS/FAIL",
"issues": [
{
"claim": "The specific claim in the resume",
"issue": "Why this is problematic",
"source_check": "Result of checking against source materials",
"suggested_correction": "How to fix this issue"
}
]
},
"skill_verification": {
"status": "PASS/FAIL",
"unverified_skills": ["skill1", "skill2"...]
},
"experience_verification": {
"status": "PASS/FAIL",
"problematic_statements": [
{
"statement": "The problematic experience statement",
"issue": "Why this is problematic",
"suggested_correction": "How to fix this issue"
}
]
},
"overall_assessment": "APPROVED/NEEDS REVISION",
"correction_instructions": "Specific instructions for correcting the resume"
}
}
```
## CRITICAL VERIFICATION CRITERIA:
1. Any skill mentioned in the resume MUST appear verbatim in the original materials
2. Any technology experience claimed MUST be explicitly documented in original materials
3. Role descriptions must not imply expertise with technologies not listed in original materials
4. "Transferable skills" must be reasonably transferable, not stretches or fabrications
5. Job titles, dates, and companies must match exactly with original materials
6. Professional summary must not imply experience with technologies from the job description that aren't in the candidate's background
## SPECIAL ATTENTION:
Pay particular attention to subtle fabrications such as:
- Vague wording that implies experience ("worked with", "familiar with", "utilized") with technologies not in original materials
- Reframing unrelated experience to falsely imply relevance to the job requirements
- Adding technologies to project descriptions that weren't mentioned in the original materials
- Exaggerating level of involvement or expertise in technologies mentioned in passing
This verification is the final safeguard against providing a resume with fabricated information. Be thorough and critical in your assessment.
</|system|>
<|job_analysis|>
[INSERT OUTPUT FROM STAGE 1 HERE]
</|job_analysis|>
<|original_materials|>
[INSERT ORIGINAL RESUME AND CONTEXT HERE]
</|original_materials|>
<|tailored_resume|>
[INSERT GENERATED RESUME FROM STAGE 2 HERE]
</|tailored_resume|>
Perform a comprehensive verification of the tailored resume against the original materials and provide a detailed report on any discrepancies or fabrications.
````
## Why Add This Third Verification Stage:
Adding this verification stage provides several key benefits:
1. **Independent verification** - A separate LLM pass dedicated solely to fact-checking helps catch issues that might have slipped through
2. **Explicit fabrication hunting** - The prompt specifically focuses on finding fabrications rather than generating content
3. **Structured feedback** - The JSON output clearly identifies specific issues that need to be addressed
4. **Automated quality control** - You can programmatically check the verification results and only proceed if the resume passes verification
5. **Documentation of reasoning** - Each flagged issue includes an explanation of why it's problematic and how to fix it
This three-stage pipeline (analyze → generate → verify) creates multiple checkpoints to prevent fabrication while still producing relevant, tailored resumes that highlight the candidate's actual transferable skills.

47
src/server.py
View File

@ -50,14 +50,18 @@ import uvicorn # type: ignore
import numpy as np # type: ignore
import umap # type: ignore
from sklearn.preprocessing import MinMaxScaler # type: ignore
# Prometheus
from prometheus_client import Summary # type: ignore
from prometheus_fastapi_instrumentator import Instrumentator # type: ignore
from prometheus_client import CollectorRegistry, Counter # type: ignore
from utils import (
rag as Rag,
tools as Tools,
Context, Conversation, Message,
Agent,
Metrics,
Tunables,
defines,
logger,
@ -242,9 +246,21 @@ class WebServer:
def __init__(self, llm, model=MODEL_NAME):
self.app = FastAPI(lifespan=self.lifespan)
Instrumentator().instrument(self.app)
Instrumentator().expose(self.app)
self.prometheus_collector = CollectorRegistry()
self.metrics = Metrics(prometheus_collector=self.prometheus_collector)
# Keep the Instrumentator instance alive
self.instrumentator = Instrumentator(registry=self.prometheus_collector)
# Instrument the FastAPI app
self.instrumentator.instrument(self.app)
# Expose the /metrics endpoint
self.instrumentator.expose(self.app, endpoint="/metrics")
self.contexts = {}
self.llm = llm
self.model = model
self.processing = False
@ -729,10 +745,22 @@ class WebServer:
json_data = json.loads(content)
logger.info("JSON parsed successfully, attempting model validation")
# Now try Pydantic validation
self.contexts[context_id] = Context.model_validate_json(content)
self.contexts[context_id].file_watcher=self.file_watcher
# Validate from JSON (no prometheus_collector or file_watcher)
context = Context.model_validate(json_data)
# Set excluded fields
context.file_watcher = self.file_watcher
context.prometheus_collector = self.prometheus_collector
# Now set context on agents manually
agent_types = [agent.agent_type for agent in context.agents]
if len(agent_types) != len(set(agent_types)):
raise ValueError("Context cannot contain multiple agents of the same agent_type")
for agent in context.agents:
agent.set_context(context)
self.contexts[context_id] = context
logger.info(f"Successfully loaded context {context_id}")
except json.JSONDecodeError as e:
logger.error(f"Invalid JSON in file: {e}")
@ -741,7 +769,7 @@ class WebServer:
import traceback
logger.error(traceback.format_exc())
# Fallback to creating a new context
self.contexts[context_id] = Context(id=context_id, file_watcher=self.file_watcher)
self.contexts[context_id] = Context(id=context_id, file_watcher=self.file_watcher, prometheus_collector=self.prometheus_collector)
return self.contexts[context_id]
@ -758,7 +786,7 @@ class WebServer:
if not context_id:
context_id = str(uuid4())
logger.info(f"Creating new context with ID: {context_id}")
context = Context(id=context_id, file_watcher=self.file_watcher)
context = Context(id=context_id, file_watcher=self.file_watcher, prometheus_collector=self.prometheus_collector)
if os.path.exists(defines.resume_doc):
context.user_resume = open(defines.resume_doc, "r").read()
@ -819,6 +847,9 @@ class WebServer:
if message.status != "done":
yield message
logger.info(f"{agent_type}.process_message: {message.status} {f'...{message.response[-20:]}' if len(message.response) > 20 else message.response}")
if message.metadata["eval_count"]:
agent.metrics.tokens_prompt.labels(agent=agent.agent_type).inc(message.metadata["prompt_eval_count"])
agent.metrics.tokens_eval.labels(agent=agent.agent_type).inc(message.metadata["eval_count"])
message.status = "done"
yield message
return

0
src/tests/__init__.py Normal file
View File

40
src/tests/test-agent.py Normal file
View File

@ -0,0 +1,40 @@
from .. utils import logger
import ollama
from .. utils import (
rag as Rag,
Context,
defines
)
import json
llm = ollama.Client(host=defines.ollama_api_url)
observer, file_watcher = Rag.start_file_watcher(
llm=llm,
watch_directory=defines.doc_dir,
recreate=False # Don't recreate if exists
)
context = Context(file_watcher=file_watcher)
data = context.model_dump(mode='json')
context = Context.model_validate_json(json.dumps(data))
context.file_watcher = file_watcher
agent = context.get_or_create_agent("chat", system_prompt="You are a helpful assistant.")
# logger.info(f"data: {data}")
# logger.info(f"agent: {agent}")
agent_type = agent.get_agent_type()
logger.info(f"agent_type: {agent_type}")
logger.info(f"system_prompt: {agent.system_prompt}")
agent.system_prompt = "Eat more tomatoes."
data = context.model_dump(mode='json')
context = Context.model_validate_json(json.dumps(data))
context.file_watcher = file_watcher
agent = context.get_agent("chat")
logger.info(f"system_prompt: {agent.system_prompt}")

109
src/tests/test-context-routing.py Normal file
View File

@ -0,0 +1,109 @@
from fastapi import FastAPI, Request, Depends, Query # type: ignore
from fastapi.responses import RedirectResponse, JSONResponse # type: ignore
from uuid import UUID, uuid4
import logging
import traceback
from typing import Callable, Optional
from anyio.to_thread import run_sync # type: ignore
logger = logging.getLogger(__name__)
class RedirectToContext(Exception):
def __init__(self, url: str):
self.url = url
logger.info(f"Redirect to Context: {url}")
super().__init__(f"Redirect to Context: {url}")
class ContextRouteManager:
def __init__(self, app: FastAPI):
self.app = app
self.setup_handlers()
def setup_handlers(self):
@self.app.exception_handler(RedirectToContext)
async def handle_context_redirect(request: Request, exc: RedirectToContext):
logger.info(f"Handling redirect to {exc.url}")
return RedirectResponse(url=exc.url, status_code=307)
def ensure_context(self, route_name: str = "context_id") -> Callable[[Request], Optional[UUID]]:
logger.info(f"Setting up context dependency for route parameter: {route_name}")
async def _ensure_context_dependency(request: Request) -> Optional[UUID]:
logger.info(f"Entering ensure_context_dependency, Request URL: {request.url}")
logger.info(f"Path params: {request.path_params}")
path_params = request.path_params
route_value = path_params.get(route_name)
logger.info(f"route_value: {route_value!r}, type: {type(route_value)}")
if route_value is None or not isinstance(route_value, str) or not route_value.strip():
logger.info(f"route_value is invalid, generating new UUID")
path = request.url.path.rstrip('/')
new_context = await run_sync(uuid4)
redirect_url = f"{path}/{new_context}"
logger.info(f"Redirecting to {redirect_url}")
raise RedirectToContext(redirect_url)
logger.info(f"Attempting to parse route_value as UUID: {route_value}")
try:
route_context = await run_sync(UUID, route_value)
logger.info(f"Successfully parsed UUID: {route_context}")
return route_context
except ValueError as e:
logger.error(f"Failed to parse UUID from route_value: {route_value!r}, error: {str(e)}")
path = request.url.path.rstrip('/')
new_context = await run_sync(uuid4)
redirect_url = f"{path}/{new_context}"
logger.info(f"Invalid UUID, redirecting to {redirect_url}")
raise RedirectToContext(redirect_url)
return _ensure_context_dependency # type: ignore
def route_pattern(self, path: str, *dependencies, **kwargs):
logger.info(f"Registering route: {path}")
ensure_context = self.ensure_context()
def decorator(func):
all_dependencies = list(dependencies)
all_dependencies.append(Depends(ensure_context))
logger.info(f"Route {path} registered with dependencies: {all_dependencies}")
return self.app.get(path, dependencies=all_dependencies, **kwargs)(func)
return decorator
app = FastAPI(redirect_slashes=True)
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
logger.error(f"Unhandled exception: {str(exc)}")
logger.error(f"Request URL: {request.url}, Path params: {request.path_params}")
logger.error(f"Stack trace: {''.join(traceback.format_tb(exc.__traceback__))}")
return JSONResponse(
status_code=500,
content={"error": "Internal server error", "detail": str(exc)}
)
@app.middleware("http")
async def log_requests(request: Request, call_next):
logger.info(f"Incoming request: {request.method} {request.url}, Path params: {request.path_params}")
response = await call_next(request)
return response
context_router = ContextRouteManager(app)
@context_router.route_pattern("/api/history/{context_id}")
async def get_history(request: Request, context_id: UUID = Depends(context_router.ensure_context()), agent_type: str = Query(..., description="Type of agent to retrieve history for")):
logger.info(f"{request.method} {request.url.path} with context_id: {context_id}")
return {"context_id": str(context_id), "agent_type": agent_type}
@app.get("/api/history")
async def redirect_history(request: Request, agent_type: str = Query(..., description="Type of agent to retrieve history for")):
path = request.url.path.rstrip('/')
new_context = uuid4()
redirect_url = f"{path}/{new_context}?agent_type={agent_type}"
logger.info(f"Redirecting from /api/history to {redirect_url}")
return RedirectResponse(url=redirect_url, status_code=307)
if __name__ == "__main__":
import uvicorn # type: ignore
uvicorn.run(app, host="0.0.0.0", port=8900)

28
src/tests/test-context.py Normal file
View File

@ -0,0 +1,28 @@
# From /opt/backstory run:
# python -m src.tests.test-context
import os
os.environ["TORCH_CPP_LOG_LEVEL"] = "ERROR"
import warnings
warnings.filterwarnings("ignore", message="Couldn't find ffmpeg or avconv")
import ollama
from .. utils import (
rag as Rag,
Context,
defines
)
import json
llm = ollama.Client(host=defines.ollama_api_url) # type: ignore
observer, file_watcher = Rag.start_file_watcher(
llm=llm,
watch_directory=defines.doc_dir,
recreate=False # Don't recreate if exists
)
context = Context(file_watcher=file_watcher)
data = context.model_dump(mode='json')
context = Context.from_json(json.dumps(data), file_watcher=file_watcher)

13
src/tests/test-message.py Normal file
View File

@ -0,0 +1,13 @@
# From /opt/backstory run:
# python -m src.tests.test-message
from .. utils import logger
from .. utils import (
Message
)
import json
prompt = "This is a test"
message = Message(prompt=prompt)
print(message.model_dump(mode='json'))

19
src/tests/test-metrics.py Normal file
View File

@ -0,0 +1,19 @@
# From /opt/backstory run:
# python -m src.tests.test-metrics
from .. utils import (
Metrics
)
import json
# Get the singleton Metrics instance
metrics = Metrics()
# Use the existing metrics
metrics.prepare_count.labels(agent="chat").inc()
metrics.prepare_duration.labels(agent="prepare").observe(0.45)
json = metrics.model_dump(mode='json')
metrics = Metrics.model_validate(json)
print(metrics)

2
src/utils/__init__.py
View File

@ -9,6 +9,7 @@ from . message import Message, Tunables
from . rag import ChromaDBFileWatcher, start_file_watcher
from . setup_logging import setup_logging
from . agents import class_registry, AnyAgent, Agent, __all__ as agents_all
from . metrics import Metrics
__all__ = [
'Agent',
@ -16,6 +17,7 @@ __all__ = [
'Context',
'Conversation',
'Message',
'Metrics',
'ChromaDBFileWatcher',
'start_file_watcher',
'logger',

4
src/utils/agents/__init__.py
View File

@ -4,14 +4,14 @@ import importlib
import pathlib
import inspect
from . types import registry
from . types import agent_registry
from .. setup_logging import setup_logging
from .. import defines
from . base import Agent
logger = setup_logging(defines.logging_level)
__all__ = [ "AnyAgent", "Agent", "registry", "class_registry" ]
__all__ = [ "AnyAgent", "Agent", "agent_registry", "class_registry" ]
# Type alias for Agent or any subclass
AnyAgent: TypeAlias = Agent # BaseModel covers Agent and subclasses

659
src/utils/agents/base.py
View File

@ -8,11 +8,11 @@ import json
import time
import inspect
from abc import ABC
import asyncio
from prometheus_client import Counter, Summary, CollectorRegistry # type: ignore
from .. setup_logging import setup_logging
from .. import defines
from .. message import Message
from .. tools import ( TickerValue, WeatherForecast, AnalyzeSite, DateTime, llm_tools ) # type: ignore -- dynamically added to __all__
logger = setup_logging()
@ -20,10 +20,12 @@ logger = setup_logging()
if TYPE_CHECKING:
from .. context import Context
from .types import registry
from .. conversation import Conversation
from . types import agent_registry
from .. import defines
from .. message import Message, Tunables
from .. metrics import Metrics
from .. tools import ( TickerValue, WeatherForecast, AnalyzeSite, DateTime, llm_tools ) # type: ignore -- dynamically added to __all__
from .. conversation import Conversation
class LLMMessage(BaseModel):
role : str = Field(default="")
@ -39,6 +41,7 @@ class Agent(BaseModel, ABC):
agent_type: Literal["base"] = "base"
_agent_type: ClassVar[str] = agent_type # Add this for registration
# Tunables (sets default for new Messages attached to this agent)
tunables: Tunables = Field(default_factory=Tunables)
@ -47,7 +50,8 @@ class Agent(BaseModel, ABC):
conversation: Conversation = Conversation()
context_tokens: int = 0
context: Optional[Context] = Field(default=None, exclude=True) # Avoid circular reference, require as param, and prevent serialization
metrics: Metrics = Field(default_factory=Metrics, exclude=True)
# context_size is shared across all subclasses
_context_size: ClassVar[int] = int(defines.max_context * 0.5)
@property
@ -92,8 +96,12 @@ class Agent(BaseModel, ABC):
super().__init_subclass__(**kwargs)
# Register this class if it has an agent_type
if hasattr(cls, 'agent_type') and cls.agent_type != Agent._agent_type:
registry.register(cls.agent_type, cls)
agent_registry.register(cls.agent_type, cls)
# def __init__(self, *, context=context, **data):
# super().__init__(**data)
# self.set_context(context)
def model_dump(self, *args, **kwargs) -> Any:
# Ensure context is always excluded, even with exclude_unset=True
kwargs.setdefault("exclude", set())
@ -108,9 +116,9 @@ class Agent(BaseModel, ABC):
"""Return the set of valid agent_type values."""
return set(get_args(cls.__annotations__["agent_type"]))
def set_context(self, context):
def set_context(self, context: Context):
object.__setattr__(self, "context", context)
# Agent methods
def get_agent_type(self):
return self._agent_type
@ -121,362 +129,377 @@ class Agent(BaseModel, ABC):
"""
logger.info(f"{self.agent_type} - {inspect.stack()[0].function}")
if not self.context:
raise ValueError("Context is not set for this agent.")
# Generate RAG content if enabled, based on the content
rag_context = ""
if message.tunables.enable_rag and message.prompt:
# Gather RAG results, yielding each result
# as it becomes available
for message in self.context.generate_rag_results(message):
logger.info(f"RAG: {message.status} - {message.response}")
if message.status == "error":
yield message
return
if message.status != "done":
yield message
if "rag" in message.metadata and message.metadata["rag"]:
for rag in message.metadata["rag"]:
for doc in rag["documents"]:
rag_context += f"{doc}\n"
self.metrics.prepare_count.labels(agent=self.agent_type).inc()
with self.metrics.prepare_duration.labels(agent=self.agent_type).time():
if not self.context:
raise ValueError("Context is not set for this agent.")
# Generate RAG content if enabled, based on the content
rag_context = ""
if message.tunables.enable_rag and message.prompt:
# Gather RAG results, yielding each result
# as it becomes available
for message in self.context.generate_rag_results(message):
logger.info(f"RAG: {message.status} - {message.response}")
if message.status == "error":
yield message
return
if message.status != "done":
yield message
if "rag" in message.metadata and message.metadata["rag"]:
for rag in message.metadata["rag"]:
for doc in rag["documents"]:
rag_context += f"{doc}\n"
message.preamble = {}
message.preamble = {}
if rag_context:
message.preamble["context"] = rag_context
if rag_context:
message.preamble["context"] = rag_context
if message.tunables.enable_context and self.context.user_resume:
message.preamble["resume"] = self.context.user_resume
message.system_prompt = self.system_prompt
message.status = "done"
yield message
if message.tunables.enable_context and self.context.user_resume:
message.preamble["resume"] = self.context.user_resume
message.system_prompt = self.system_prompt
message.status = "done"
yield message
return
async def process_tool_calls(self, llm: Any, model: str, message: Message, tool_message: Any, messages: List[LLMMessage]) -> AsyncGenerator[Message, None]:
logger.info(f"{self.agent_type} - {inspect.stack()[0].function}")
if not self.context:
raise ValueError("Context is not set for this agent.")
if not message.metadata["tools"]:
raise ValueError("tools field not initialized")
tool_metadata = message.metadata["tools"]
tool_metadata["tool_calls"] = []
message.status = "tooling"
for i, tool_call in enumerate(tool_message.tool_calls):
arguments = tool_call.function.arguments
tool = tool_call.function.name
# Yield status update before processing each tool
message.response = f"Processing tool {i+1}/{len(tool_message.tool_calls)}: {tool}..."
yield message
logger.info(f"LLM - {message.response}")
self.metrics.tool_count.labels(agent=self.agent_type).inc()
with self.metrics.tool_duration.labels(agent=self.agent_type).time():
if not self.context:
raise ValueError("Context is not set for this agent.")
if not message.metadata["tools"]:
raise ValueError("tools field not initialized")
tool_metadata = message.metadata["tools"]
tool_metadata["tool_calls"] = []
message.status = "tooling"
for i, tool_call in enumerate(tool_message.tool_calls):
arguments = tool_call.function.arguments
tool = tool_call.function.name
# Process the tool based on its type
match tool:
case "TickerValue":
ticker = arguments.get("ticker")
if not ticker:
# Yield status update before processing each tool
message.response = f"Processing tool {i+1}/{len(tool_message.tool_calls)}: {tool}..."
yield message
logger.info(f"LLM - {message.response}")
# Process the tool based on its type
match tool:
case "TickerValue":
ticker = arguments.get("ticker")
if not ticker:
ret = None
else:
ret = TickerValue(ticker)
case "AnalyzeSite":
url = arguments.get("url")
question = arguments.get("question", "what is the summary of this content?")
# Additional status update for long-running operations
message.response = f"Retrieving and summarizing content from {url}..."
yield message
ret = await AnalyzeSite(llm=llm, model=model, url=url, question=question)
case "DateTime":
tz = arguments.get("timezone")
ret = DateTime(tz)
case "WeatherForecast":
city = arguments.get("city")
state = arguments.get("state")
message.response = f"Fetching weather data for {city}, {state}..."
yield message
ret = WeatherForecast(city, state)
case _:
ret = None
else:
ret = TickerValue(ticker)
case "AnalyzeSite":
url = arguments.get("url")
question = arguments.get("question", "what is the summary of this content?")
# Build response for this tool
tool_response = {
"role": "tool",
"content": json.dumps(ret),
"name": tool_call.function.name
}
tool_metadata["tool_calls"].append(tool_response)
# Additional status update for long-running operations
message.response = f"Retrieving and summarizing content from {url}..."
if len(tool_metadata["tool_calls"]) == 0:
message.status = "done"
yield message
ret = await AnalyzeSite(llm=llm, model=model, url=url, question=question)
case "DateTime":
tz = arguments.get("timezone")
ret = DateTime(tz)
case "WeatherForecast":
city = arguments.get("city")
state = arguments.get("state")
return
message.response = f"Fetching weather data for {city}, {state}..."
yield message
ret = WeatherForecast(city, state)
case _:
ret = None
# Build response for this tool
tool_response = {
"role": "tool",
"content": json.dumps(ret),
"name": tool_call.function.name
}
message_dict = LLMMessage(
role=tool_message.get("role", "assistant"),
content=tool_message.get("content", ""),
tool_calls=[ {
"function": {
"name": tc["function"]["name"],
"arguments": tc["function"]["arguments"]
}
} for tc in tool_message.tool_calls
]
)
tool_metadata["tool_calls"].append(tool_response)
if len(tool_metadata["tool_calls"]) == 0:
message.status = "done"
yield message
return
messages.append(message_dict)
messages.extend(tool_metadata["tool_calls"])
message_dict = LLMMessage(
role=tool_message.get("role", "assistant"),
content=tool_message.get("content", ""),
tool_calls=[ {
"function": {
"name": tc["function"]["name"],
"arguments": tc["function"]["arguments"]
}
} for tc in tool_message.tool_calls
]
)
message.status = "thinking"
message.response = "Incorporating tool results into response..."
yield message
messages.append(message_dict)
messages.extend(tool_metadata["tool_calls"])
# Decrease creativity when processing tool call requests
message.response = ""
start_time = time.perf_counter()
for response in llm.chat(
model=model,
messages=messages,
stream=True,
options={
**message.metadata["options"],
# "temperature": 0.5,
}
):
# logger.info(f"LLM::Tools: {'done' if response.done else 'processing'} - {response.message}")
message.status = "streaming"
message.response += response.message.content
if not response.done:
yield message
if response.done:
message.metadata["eval_count"] += response.eval_count
message.metadata["eval_duration"] += response.eval_duration
message.metadata["prompt_eval_count"] += response.prompt_eval_count
message.metadata["prompt_eval_duration"] += response.prompt_eval_duration
self.context_tokens = response.prompt_eval_count + response.eval_count
message.status = "done"
yield message
message.status = "thinking"
message.response = "Incorporating tool results into response..."
yield message
# Decrease creativity when processing tool call requests
message.response = ""
start_time = time.perf_counter()
for response in llm.chat(
model=model,
messages=messages,
stream=True,
options={
**message.metadata["options"],
# "temperature": 0.5,
}
):
# logger.info(f"LLM::Tools: {'done' if response.done else 'processing'} - {response.message}")
message.status = "streaming"
message.response += response.message.content
if not response.done:
yield message
if response.done:
message.metadata["eval_count"] += response.eval_count
message.metadata["eval_duration"] += response.eval_duration
message.metadata["prompt_eval_count"] += response.prompt_eval_count
message.metadata["prompt_eval_duration"] += response.prompt_eval_duration
self.context_tokens = response.prompt_eval_count + response.eval_count
message.status = "done"
yield message
end_time = time.perf_counter()
message.metadata["timers"]["llm_with_tools"] = f"{(end_time - start_time):.4f}"
return
end_time = time.perf_counter()
message.metadata["timers"]["llm_with_tools"] = f"{(end_time - start_time):.4f}"
return
async def generate_llm_response(self, llm: Any, model: str, message: Message) -> AsyncGenerator[Message, None]:
logger.info(f"{self.agent_type} - {inspect.stack()[0].function}")
self.metrics.generate_count.labels(agent=self.agent_type).inc()
with self.metrics.generate_duration.labels(agent=self.agent_type).time():
if not self.context:
raise ValueError("Context is not set for this agent.")
if not self.context:
raise ValueError("Context is not set for this agent.")
# Create a pruned down message list based purely on the prompt and responses,
# discarding the full preamble generated by prepare_message
messages: List[LLMMessage] = [ LLMMessage(role="system", content=message.system_prompt) ]
messages.extend([
item for m in self.conversation
for item in [
LLMMessage(role="user", content=m.prompt.strip()),
LLMMessage(role="assistant", content=m.response.strip())
]
])
# Only the actual user query is provided with the full context message
messages.append(LLMMessage(role="user", content=message.context_prompt.strip()))
# Create a pruned down message list based purely on the prompt and responses,
# discarding the full preamble generated by prepare_message
messages: List[LLMMessage] = [ LLMMessage(role="system", content=message.system_prompt) ]
messages.extend([
item for m in self.conversation
for item in [
LLMMessage(role="user", content=m.prompt.strip()),
LLMMessage(role="assistant", content=m.response.strip())
]
])
# Only the actual user query is provided with the full context message
messages.append(LLMMessage(role="user", content=message.context_prompt.strip()))
#message.metadata["messages"] = messages
message.metadata["options"]={
"seed": 8911,
"num_ctx": self.context_size,
#"temperature": 0.9, # Higher temperature to encourage tool usage
}
#message.metadata["messages"] = messages
message.metadata["options"]={
"seed": 8911,
"num_ctx": self.context_size,
#"temperature": 0.9, # Higher temperature to encourage tool usage
}
# Create a dict for storing various timing stats
message.metadata["timers"] = {}
# Create a dict for storing various timing stats
message.metadata["timers"] = {}
use_tools = message.tunables.enable_tools and len(self.context.tools) > 0
message.metadata["tools"] = {
"available": llm_tools(self.context.tools),
"used": False
}
tool_metadata = message.metadata["tools"]
use_tools = message.tunables.enable_tools and len(self.context.tools) > 0
message.metadata["tools"] = {
"available": llm_tools(self.context.tools),
"used": False
}
tool_metadata = message.metadata["tools"]
if use_tools:
message.status = "thinking"
message.response = f"Performing tool analysis step 1/2..."
yield message
if use_tools:
logger.info("Checking for LLM tool usage")
start_time = time.perf_counter()
# Tools are enabled and available, so query the LLM with a short context of messages
# in case the LLM did something like ask "Do you want me to run the tool?" and the
# user said "Yes" -- need to keep the context in the thread.
tool_metadata["messages"] = (
[{"role": "system", "content": self.system_prompt}] + messages[-6:]
if len(messages) >= 7
else messages
)
response = llm.chat(
model=model,
messages=tool_metadata["messages"],
tools=tool_metadata["available"],
options={
**message.metadata["options"],
#"num_predict": 1024, # "Low" token limit to cut off after tool call
},
stream=False # No need to stream the probe
)
end_time = time.perf_counter()
message.metadata["timers"]["tool_check"] = f"{(end_time - start_time):.4f}"
if not response.message.tool_calls:
logger.info("LLM indicates tools will not be used")
# The LLM will not use tools, so disable use_tools so we can stream the full response
use_tools = False
else:
tool_metadata["attempted"] = response.message.tool_calls
if use_tools:
logger.info("LLM indicates tools will be used")
# Tools are enabled and available and the LLM indicated it will use them
message.response = f"Performing tool analysis step 2/2 (tool use suspected)..."
yield message
logger.info(f"Performing LLM call with tools")
start_time = time.perf_counter()
response = llm.chat(
model=model,
messages=tool_metadata["messages"], # messages,
tools=tool_metadata["available"],
options={
**message.metadata["options"],
},
stream=False
)
end_time = time.perf_counter()
message.metadata["timers"]["non_streaming"] = f"{(end_time - start_time):.4f}"
if not response:
message.status = "error"
message.response = "No response from LLM."
yield message
return
if response.message.tool_calls:
tool_metadata["used"] = response.message.tool_calls
# Process all yielded items from the handler
start_time = time.perf_counter()
async for message in self.process_tool_calls(llm=llm, model=model, message=message, tool_message=response.message, messages=messages):
if message.status == "error":
yield message
return
yield message
end_time = time.perf_counter()
message.metadata["timers"]["process_tool_calls"] = f"{(end_time - start_time):.4f}"
message.status = "done"
return
logger.info("LLM indicated tools will be used, and then they weren't")
message.response = response.message.content
message.status = "done"
yield message
return
# not use_tools
message.status = "thinking"
message.response = f"Performing tool analysis step 1/2..."
message.response = f"Generating response..."
yield message
logger.info("Checking for LLM tool usage")
# Reset the response for streaming
message.response = ""
start_time = time.perf_counter()
# Tools are enabled and available, so query the LLM with a short context of messages
# in case the LLM did something like ask "Do you want me to run the tool?" and the
# user said "Yes" -- need to keep the context in the thread.
tool_metadata["messages"] = (
[{"role": "system", "content": self.system_prompt}] + messages[-6:]
if len(messages) >= 7
else messages
)
response = llm.chat(
model=model,
messages=tool_metadata["messages"],
tools=tool_metadata["available"],
for response in llm.chat(
model=model,
messages=messages,
options={
**message.metadata["options"],
#"num_predict": 1024, # "Low" token limit to cut off after tool call
},
stream=False # No need to stream the probe
)
end_time = time.perf_counter()
message.metadata["timers"]["tool_check"] = f"{(end_time - start_time):.4f}"
if not response.message.tool_calls:
logger.info("LLM indicates tools will not be used")
# The LLM will not use tools, so disable use_tools so we can stream the full response
use_tools = False
else:
tool_metadata["attempted"] = response.message.tool_calls
if use_tools:
logger.info("LLM indicates tools will be used")
# Tools are enabled and available and the LLM indicated it will use them
message.response = f"Performing tool analysis step 2/2 (tool use suspected)..."
yield message
logger.info(f"Performing LLM call with tools")
start_time = time.perf_counter()
response = llm.chat(
model=model,
messages=tool_metadata["messages"], # messages,
tools=tool_metadata["available"],
options={
**message.metadata["options"],
},
stream=False
)
end_time = time.perf_counter()
message.metadata["timers"]["non_streaming"] = f"{(end_time - start_time):.4f}"
if not response:
message.status = "error"
message.response = "No response from LLM."
yield message
return
if response.message.tool_calls:
tool_metadata["used"] = response.message.tool_calls
# Process all yielded items from the handler
start_time = time.perf_counter()
async for message in self.process_tool_calls(llm=llm, model=model, message=message, tool_message=response.message, messages=messages):
if message.status == "error":
yield message
return
stream=True,
):
if not response:
message.status = "error"
message.response = "No response from LLM."
yield message
end_time = time.perf_counter()
message.metadata["timers"]["process_tool_calls"] = f"{(end_time - start_time):.4f}"
message.status = "done"
return
logger.info("LLM indicated tools will be used, and then they weren't")
message.response = response.message.content
message.status = "done"
yield message
return
return
# not use_tools
message.status = "thinking"
message.response = f"Generating response..."
yield message
# Reset the response for streaming
message.response = ""
start_time = time.perf_counter()
for response in llm.chat(
model=model,
messages=messages,
options={
**message.metadata["options"],
},
stream=True,
):
if not response:
message.status = "error"
message.response = "No response from LLM."
yield message
return
message.status = "streaming"
message.response += response.message.content
message.status = "streaming"
message.response += response.message.content
if not response.done:
yield message
if not response.done:
yield message
if response.done:
message.metadata["eval_count"] += response.eval_count
message.metadata["eval_duration"] += response.eval_duration
message.metadata["prompt_eval_count"] += response.prompt_eval_count
message.metadata["prompt_eval_duration"] += response.prompt_eval_duration
self.context_tokens = response.prompt_eval_count + response.eval_count
message.status = "done"
yield message
if response.done:
message.metadata["eval_count"] += response.eval_count
message.metadata["eval_duration"] += response.eval_duration
message.metadata["prompt_eval_count"] += response.prompt_eval_count
message.metadata["prompt_eval_duration"] += response.prompt_eval_duration
self.context_tokens = response.prompt_eval_count + response.eval_count
message.status = "done"
yield message
end_time = time.perf_counter()
message.metadata["timers"]["streamed"] = f"{(end_time - start_time):.4f}"
end_time = time.perf_counter()
message.metadata["timers"]["streamed"] = f"{(end_time - start_time):.4f}"
return
async def process_message(self, llm: Any, model: str, message:Message) -> AsyncGenerator[Message, None]:
logger.info(f"{self.agent_type} - {inspect.stack()[0].function}")
if not self.context:
raise ValueError("Context is not set for this agent.")
self.metrics.process_count.labels(agent=self.agent_type).inc()
with self.metrics.process_duration.labels(agent=self.agent_type).time():
if not self.context:
raise ValueError("Context is not set for this agent.")
if self.context.processing:
logger.info("TODO: Implement delay queing; busy for same agent, otherwise return queue size and estimated wait time")
message.status = "error"
message.response = "Busy processing another request."
yield message
return
self.context.processing = True
message.metadata["system_prompt"] = f"<|system|>\n{self.system_prompt.strip()}\n"
message.context_prompt = ""
for p in message.preamble.keys():
message.context_prompt += f"\n<|{p}|>\n{message.preamble[p].strip()}\n"
message.context_prompt += f"{message.prompt}"
# Estimate token length of new messages
message.response = f"Optimizing context..."
message.status = "thinking"
yield message
message.metadata["context_size"] = self.set_optimal_context_size(llm, model, prompt=message.context_prompt)
message.response = f"Processing {'RAG augmented ' if message.metadata['rag'] else ''}query..."
message.status = "thinking"
yield message
async for message in self.generate_llm_response(llm, model, message):
# logger.info(f"LLM: {message.status} - {f'...{message.response[-20:]}' if len(message.response) > 20 else message.response}")
if message.status == "error":
spinner: List[str] = ['\\', '|', '/', '-']
tick : int = 0
while self.context.processing:
message.status = "waiting"
message.response = f"Busy processing another request. Please wait. {spinner[tick]}"
tick = (tick + 1) % len(spinner)
yield message
self.context.processing = False
return
await asyncio.sleep(1) # Allow the event loop to process the write
self.context.processing = True
message.metadata["system_prompt"] = f"<|system|>\n{self.system_prompt.strip()}\n</|system|>"
message.context_prompt = ""
for p in message.preamble.keys():
message.context_prompt += f"\n<|{p}|>\n{message.preamble[p].strip()}\n</|{p}>\n\n"
message.context_prompt += f"{message.prompt}"
# Estimate token length of new messages
message.response = f"Optimizing context..."
message.status = "thinking"
yield message
# Done processing, add message to conversation
message.status = "done"
self.conversation.add(message)
self.context.processing = False
message.metadata["context_size"] = self.set_optimal_context_size(llm, model, prompt=message.context_prompt)
message.response = f"Processing {'RAG augmented ' if message.metadata['rag'] else ''}query..."
message.status = "thinking"
yield message
async for message in self.generate_llm_response(llm, model, message):
# logger.info(f"LLM: {message.status} - {f'...{message.response[-20:]}' if len(message.response) > 20 else message.response}")
if message.status == "error":
yield message
self.context.processing = False
return
yield message
# Done processing, add message to conversation
message.status = "done"
self.conversation.add(message)
self.context.processing = False
return
# Register the base agent
registry.register(Agent._agent_type, Agent)
agent_registry.register(Agent._agent_type, Agent)

7
src/utils/agents/chat.py
View File

@ -3,7 +3,7 @@ from typing import Literal, AsyncGenerator, ClassVar, Optional, Any
from datetime import datetime
import inspect
from . base import Agent, registry
from . base import Agent, agent_registry
from .. message import Message
from .. setup_logging import setup_logging
logger = setup_logging()
@ -45,7 +45,8 @@ class Chat(Agent):
yield message
if message.preamble:
preamble_types = [f"<|{p}|>" for p in message.preamble.keys()]
excluded = {}
preamble_types = [f"<|{p}|>" for p in message.preamble.keys() if p not in excluded]
preamble_types_AND = " and ".join(preamble_types)
preamble_types_OR = " or ".join(preamble_types)
message.preamble["rules"] = f"""\
@ -56,4 +57,4 @@ class Chat(Agent):
message.preamble["question"] = "Respond to:"
# Register the base agent
registry.register(Chat._agent_type, Chat)
agent_registry.register(Chat._agent_type, Chat)

9
src/utils/agents/fact_check.py
View File

@ -4,7 +4,7 @@ from typing import Literal, ClassVar, Optional, Any, AsyncGenerator, List # NOTE
from datetime import datetime
import inspect
from . base import Agent, registry
from . base import Agent, agent_registry
from .. conversation import Conversation
from .. message import Message
from .. setup_logging import setup_logging
@ -43,7 +43,7 @@ class FactCheck(Agent):
if not resume_agent:
raise ValueError("resume agent does not exist")
message.enable_tools = False
message.tunables.enable_tools = False
async for message in super().prepare_message(message):
if message.status != "done":
@ -52,7 +52,8 @@ class FactCheck(Agent):
message.preamble["generated-resume"] = resume_agent.resume
message.preamble["discrepancies"] = self.facts
preamble_types = [f"<|{p}|>" for p in message.preamble.keys()]
excluded = {"job_description"}
preamble_types = [f"<|{p}|>" for p in message.preamble.keys() if p not in excluded]
preamble_types_AND = " and ".join(preamble_types)
preamble_types_OR = " or ".join(preamble_types)
message.preamble["rules"] = f"""\
@ -66,4 +67,4 @@ class FactCheck(Agent):
return
# Register the base agent
registry.register(FactCheck._agent_type, FactCheck)
agent_registry.register(FactCheck._agent_type, FactCheck)

140
src/utils/agents/job_description.py
View File

@ -4,40 +4,83 @@ from typing import Literal, ClassVar, Optional, Any, AsyncGenerator, List # NOTE
from datetime import datetime
import inspect
from . base import Agent, registry
from . base import Agent, agent_registry
from .. conversation import Conversation
from .. message import Message
from .. setup_logging import setup_logging
logger = setup_logging()
system_generate_resume = f"""
Launched on {datetime.now().isoformat()}.
system_generate_resume = """
You are an objective skills analyzer for resume tailoring. Your task is to
analyze a job description and a candidate's background, identifying ONLY
legitimate matches without fabrication. Your analysis will be used in a
second stage to generate a tailored resume.
You are a professional resume writer. Your task is to write a concise, polished, and tailored resume for a specific job based only on the individual's <|context|>.
## INSTRUCTIONS:
When answering queries, follow these steps:
1. Analyze the job description to extract:
- Required skills and technologies
- Required experience types
- Company values and culture indicators
- Primary responsibilities of the role
- You must not invent or assume any inforation not explicitly present in the <|context|>.
- Analyze the <|job_description|> to identify skills required for the job.
- Use the <|job_description|> provided to guide the focus, tone, and relevant skills or experience to highlight from the <|context|>.
- Identify and emphasize the experiences, achievements, and responsibilities from the <|context|> that best align with the <|job_description|>.
- Only provide information from <|context|> items if it is relevant to the <|job_description|>.
- Do not use the <|job_description|> skills unless listed in <|context|>.
- Do not include any information unless it is provided in <|context|>.
- Use the <|context|> to create a polished, professional resume.
- Do not list any locations or mailing addresses in the resume.
- If there is information in the <|context|>, <|job_description|>, <|context|>, or <|resume|> sections to enhance the answer, incorporate it seamlessly and refer to it using natural language instead of mentioning '<|job_description|>' (etc.) or quoting it directly.
- Avoid phrases like 'According to the <|context|>' or similar references to the <|context|>, <|job_description|>, or <|context|> tags.
- Ensure the langauge is clear, concise, and aligned with industry standards for professional resumes.
2. Analyze the candidate's resume and context to identify:
- Actual skills and technologies the candidate possesses
- Actual experience types the candidate has
- Relevant achievements that align with the job
Structure the resume professionally with the following sections where applicable:
3. Create a structured output with these sections:
* Name: Use full name
* Professional Summary: A 2-4 sentence overview tailored to the job.
* Skills: A bullet list of key skills derived from the work history and relevant to the job.
* Professional Experience: A detailed list of roles, achievements, and responsibilities from <|context|> that relate to the <|job_description|>.
* Education: Include only if available in the work history.
* Notes: Indicate the initial draft of the resume was generated using the Backstory application.
### OUTPUT FORMAT:
```json
{
"job_requirements": {
"required_skills": ["skill1", "skill2"...],
"preferred_skills": ["skill1", "skill2"...],
"required_experience": ["exp1", "exp2"...],
"company_values": ["value1", "value2"...]
},
"candidate_qualifications": {
"matching_skills": ["skill1", "skill2"...],
"transferable_skills": ["skill1", "skill2"...],
"missing_skills": ["skill1", "skill2"...],
"relevant_experience": [
{
"role": "Role title",
"relevance": "Explanation of how this role is relevant"
}
],
"relevant_achievements": [
{
"achievement": "Achievement description",
"relevance": "Explanation of relevance to job"
}
]
},
"emphasis_recommendations": {
"highlight_skills": ["skill1", "skill2"...],
"highlight_experiences": ["exp1", "exp2"...],
"transferable_narratives": [
"Description of how candidate's experience in X transfers to required skill Y"
],
"honest_gap_acknowledgment": [
"Skills/technologies the candidate lacks with no reasonable transfer claim"
]
}
}
```
### CRITICAL RULES:
1. DO NOT INVENT OR IMPLY ANY SKILLS OR EXPERIENCE not explicitly present in the candidate's materials
2. A skill is only "matching" if the EXACT SAME technology/skill appears in both job requirements and candidate background
3. "Transferable skills" must have a legitimate connection - don't stretch credibility
4. Be brutally honest in "missing_skills" and "honest_gap_acknowledgment" sections
5. For each skill or experience match, note the specific section/line from the candidate's materials where it appears
6. If a job requirement has no legitimate match or transfer in the candidate's background, it MUST appear in "missing_skills"
The output of this analysis will be used to create a truthful, tailored resume. Accuracy is critical.
""".strip()
system_job_description = f"""
@ -52,6 +95,33 @@ When answering queries, follow these steps:
- Avoid phrases like 'According to the <|context|>' or similar references to the <|context|>, <|job_description|>, <|resume|>, or <|context|> tags.
""".strip()
# def run_resume_pipeline(job_description, resume, context):
# # Stage 1: Analysis
# analysis_prompt = create_analysis_prompt(job_description, resume, context)
# job_analysis = call_llm_with_prompt(analysis_prompt)
# # Validate analysis output
# validated_analysis = validate_job_analysis(job_analysis)
# if not validated_analysis:
# return {"error": "Analysis stage failed validation"}
# # Stage 2: Generation
# generation_prompt = create_generation_prompt(validated_analysis, resume)
# tailored_resume = call_llm_with_prompt(generation_prompt)
# # Stage 3: Verification
# verification_prompt = create_verification_prompt(validated_analysis, resume, context, tailored_resume)
# verification_result = call_llm_with_prompt(verification_prompt)
# # Process verification results
# if verification_result.get("verification_results", {}).get("overall_assessment") == "APPROVED":
# return {"status": "success", "resume": tailored_resume}
# else:
# # Optional: Implement correction loop
# corrected_resume = apply_corrections(tailored_resume, verification_result)
# return {"status": "corrected", "resume": corrected_resume}
class JobDescription(Agent):
agent_type: Literal["job_description"] = "job_description" # type: ignore
_agent_type: ClassVar[str] = agent_type # Add this for registration
@ -78,20 +148,23 @@ class JobDescription(Agent):
message.preamble["job_description"] = self.job_description
message.preamble["resume"] = self.context.user_resume
preamble_types = [f"<|{p}|>" for p in message.preamble.keys()]
excluded = {"job_description"}
preamble_types = [f"<|{p}|>" for p in message.preamble.keys() if p not in excluded]
preamble_types_AND = " and ".join(preamble_types)
preamble_types_OR = " or ".join(preamble_types)
message.preamble["rules"] = f"""\
- Answer the question based on the information provided in the {preamble_types_AND} sections by incorporating it seamlessly and refer to it using natural language instead of mentioning {preamble_types_OR} or quoting it directly.
- If there is no information in these sections, answer based on your knowledge, or use any available tools.
- Avoid phrases like 'According to the {preamble_types[0]}' or similar references to the {preamble_types_OR}.
"""
# message.preamble["rules"] = f"""\
# - Create your response based on the information provided in the {preamble_types_AND} sections by incorporating it seamlessly and refer to it using natural language instead of mentioning {preamble_types_OR} or quoting it directly.
# - If there is no information in these sections, answer based on your knowledge, or use any available tools.
# - Avoid phrases like 'According to the {preamble_types[0]}' or similar references to the {preamble_types_OR}.
# """
resume_agent = self.context.get_agent(agent_type="resume")
if resume_agent:
message.preamble["question"] = "Respond to:"
else:
message.preamble["question"] = "Generate a resume given the <|resume|> and <|job_description|>."
# message.preamble["question"] = "Write a professional resume for the <|job_description|>:"
# message.prompt = "Following the <|rules|>, generate a resume given the <|context|>, <|resume|> and <|job_description|>."
message.prompt = ""
yield message
return
@ -101,6 +174,9 @@ class JobDescription(Agent):
if not self.context:
raise ValueError("Context is not set for this agent.")
# Generating a resume should not use any tools
message.tunables.enable_tools = False
async for message in super().process_message(llm, model, message):
if message.status != "done":
yield message
@ -124,4 +200,4 @@ class JobDescription(Agent):
return
# Register the base agent
registry.register(JobDescription._agent_type, JobDescription)
agent_registry.register(JobDescription._agent_type, JobDescription)

10
src/utils/agents/resume.py
View File

@ -4,7 +4,7 @@ from typing import Literal, ClassVar, Optional, Any, AsyncGenerator, List # NOTE
from datetime import datetime
import inspect
from . base import Agent, registry
from . base import Agent, agent_registry
from .. message import Message
from .. setup_logging import setup_logging
logger = setup_logging()
@ -54,6 +54,9 @@ class Resume(Agent):
if not self.context:
raise ValueError("Context is not set for this agent.")
# Generating fact check or resume should not use any tools
message.tunables.enable_tools = False
async for message in super().prepare_message(message):
if message.status != "done":
yield message
@ -65,7 +68,8 @@ class Resume(Agent):
message.preamble["job_description"] = job_description_agent.job_description
preamble_types = [f"<|{p}|>" for p in message.preamble.keys()]
excluded = {}
preamble_types = [f"<|{p}|>" for p in message.preamble.keys() if p not in excluded]
preamble_types_AND = " and ".join(preamble_types)
preamble_types_OR = " or ".join(preamble_types)
message.preamble["rules"] = f"""\
@ -110,4 +114,4 @@ class Resume(Agent):
return
# Register the base agent
registry.register(Resume._agent_type, Resume)
agent_registry.register(Resume._agent_type, Resume)

2
src/utils/agents/types.py
View File

@ -28,4 +28,4 @@ class AgentRegistry:
return cls._registry.copy()
# Create a singleton instance
registry = AgentRegistry()
agent_registry = AgentRegistry()

16
src/utils/context.py
View File

@ -1,11 +1,12 @@
from __future__ import annotations
from pydantic import BaseModel, Field, model_validator# type: ignore
from uuid import uuid4
from typing import List, Optional, Generator
from typing import List, Optional, Generator, ClassVar, Any
from typing_extensions import Annotated, Union
import numpy as np # type: ignore
import logging
from uuid import uuid4
from prometheus_client import CollectorRegistry, Counter # type: ignore
from . message import Message, Tunables
from . rag import ChromaDBFileWatcher
@ -20,6 +21,7 @@ class Context(BaseModel):
model_config = {"arbitrary_types_allowed": True} # Allow ChromaDBFileWatcher
# Required fields
file_watcher: Optional[ChromaDBFileWatcher] = Field(default=None, exclude=True)
prometheus_collector: Optional[CollectorRegistry] = Field(default=None, exclude=True)
# Optional fields
id: str = Field(
@ -36,7 +38,7 @@ class Context(BaseModel):
agents: List[Annotated[Union[*Agent.__subclasses__()], Field(discriminator="agent_type")]] = Field( # type: ignore
default_factory=list
)
processing: bool = Field(default=False, exclude=True)
# @model_validator(mode="before")
@ -52,10 +54,10 @@ class Context(BaseModel):
agent_types = [agent.agent_type for agent in self.agents]
if len(agent_types) != len(set(agent_types)):
raise ValueError("Context cannot contain multiple agents of the same agent_type")
for agent in self.agents:
agent.set_context(self)
# for agent in self.agents:
# agent.set_context(self)
return self
def generate_rag_results(self, message: Message) -> Generator[Message, None, None]:
"""
Generate RAG results for the given query.
@ -143,7 +145,9 @@ class Context(BaseModel):
for agent_cls in Agent.__subclasses__():
if agent_cls.model_fields["agent_type"].default == agent_type:
# Create the agent instance with provided kwargs
agent = agent_cls(agent_type=agent_type, context=self, **kwargs)
agent = agent_cls(agent_type=agent_type, **kwargs)
# set_context after constructor to initialize any non-serialized data
agent.set_context(self)
self.agents.append(agent)
return agent

93
src/utils/metrics.py Normal file
View File

@ -0,0 +1,93 @@
from prometheus_client import Counter, Gauge, Summary, Histogram, Info, Enum, CollectorRegistry # type: ignore
from threading import Lock
def singleton(cls):
instance = None
lock = Lock()
def get_instance(*args, **kwargs):
nonlocal instance
with lock:
if instance is None:
instance = cls(*args, **kwargs)
return instance
return get_instance
@singleton
class Metrics():
def __init__(self, *args, prometheus_collector, **kwargs):
super().__init__(*args, **kwargs)
self.prometheus_collector = prometheus_collector
self.prepare_count : Counter = Counter(
name="prepare_total",
documentation="Total messages prepared by agent type",
labelnames=("agent",),
registry=self.prometheus_collector
)
self.prepare_duration : Histogram = Histogram(
name="prepare_duration",
documentation="Preparation duration by agent type",
labelnames=("agent",),
registry=self.prometheus_collector
)
self.process_count : Counter = Counter(
name="process",
documentation="Total messages processed by agent type",
labelnames=("agent",),
registry=self.prometheus_collector
)
self.process_duration : Histogram = Histogram(
name="process_duration",
documentation="Processing duration by agent type",
labelnames=("agent",),
registry=self.prometheus_collector
)
self.tool_count : Counter = Counter(
name="tool_total",
documentation="Total messages tooled by agent type",
labelnames=("agent",),
registry=self.prometheus_collector
)
self.tool_duration : Histogram = Histogram(
name="tool_duration",
documentation="Tool duration by agent type",
buckets=(0.1, 0.5, 1.0, 2.0, float('inf')),
labelnames=("agent",),
registry=self.prometheus_collector
)
self.generate_count : Counter = Counter(
name="generate_total",
documentation="Total messages generated by agent type",
labelnames=("agent",),
registry=self.prometheus_collector
)
self.generate_duration : Histogram = Histogram(
name="generate_duration",
documentation="Generate duration by agent type",
buckets=(0.1, 0.5, 1.0, 2.0, float('inf')),
labelnames=("agent",),
registry=self.prometheus_collector
)
self.tokens_prompt : Counter = Counter(
name="tokens_prompt",
documentation="Total tokens passed as prompt to LLM",
labelnames=("agent",),
registry=self.prometheus_collector
)
self.tokens_eval : Counter = Counter(
name="tokens_eval",
documentation="Total tokens returned by LLM",
labelnames=("agent",),
registry=self.prometheus_collector
)

14
src/utils/tools/__init__.py Normal file
View File

@ -0,0 +1,14 @@
import importlib
from . basetools import tools, llm_tools, enabled_tools, tool_functions
from .. setup_logging import setup_logging
from .. import defines
logger = setup_logging(level=defines.logging_level)
# Dynamically import all names from basetools listed in tools_all
module = importlib.import_module('.basetools', package=__package__)
for name in tool_functions:
globals()[name] = getattr(module, name)
__all__ = [ 'tools', 'llm_tools', 'enabled_tools', 'tool_functions' ]

439
src/utils/tools/basetools.py Normal file
View File

@ -0,0 +1,439 @@
import os
from datetime import datetime
from typing import (
Any,
)
from typing_extensions import Annotated
from bs4 import BeautifulSoup # type: ignore
from geopy.geocoders import Nominatim # type: ignore
import pytz # type: ignore
import requests
import yfinance as yf # type: ignore
import logging
# %%
def WeatherForecast(city, state, country="USA"):
"""
Get weather information from weather.gov based on city, state, and country.
Args:
city (str): City name
state (str): State name or abbreviation
country (str): Country name (defaults to "USA" as weather.gov is for US locations)
Returns:
dict: Weather forecast information
"""
# Step 1: Get coordinates for the location using geocoding
location = f"{city}, {state}, {country}"
coordinates = get_coordinates(location)
if not coordinates:
return {"error": f"Could not find coordinates for {location}"}
# Step 2: Get the forecast grid endpoint for the coordinates
grid_endpoint = get_grid_endpoint(coordinates)
if not grid_endpoint:
return {"error": f"Could not find weather grid for coordinates {coordinates}"}
# Step 3: Get the forecast data from the grid endpoint
forecast = get_forecast(grid_endpoint)
if not forecast['location']:
forecast['location'] = location
return forecast
def get_coordinates(location):
"""Convert a location string to latitude and longitude using Nominatim geocoder."""
try:
# Create a geocoder with a meaningful user agent
geolocator = Nominatim(user_agent="weather_app_example")
# Get the location
location_data = geolocator.geocode(location)
if location_data:
return {
"latitude": location_data.latitude,
"longitude": location_data.longitude
}
else:
print(f"Location not found: {location}")
return None
except Exception as e:
print(f"Error getting coordinates: {e}")
return None
def get_grid_endpoint(coordinates):
"""Get the grid endpoint from weather.gov based on coordinates."""
try:
lat = coordinates["latitude"]
lon = coordinates["longitude"]
# Define headers for the API request
headers = {
"User-Agent": "WeatherAppExample/1.0 (your_email@example.com)",
"Accept": "application/geo+json"
}
# Make the request to get the grid endpoint
url = f"https://api.weather.gov/points/{lat},{lon}"
response = requests.get(url, headers=headers)
if response.status_code == 200:
data = response.json()
return data["properties"]["forecast"]
else:
print(f"Error getting grid: {response.status_code} - {response.text}")
return None
except Exception as e:
print(f"Error in get_grid_endpoint: {e}")
return None
# Weather related function
def get_forecast(grid_endpoint):
"""Get the forecast data from the grid endpoint."""
try:
# Define headers for the API request
headers = {
"User-Agent": "WeatherAppExample/1.0 (your_email@example.com)",
"Accept": "application/geo+json"
}
# Make the request to get the forecast
response = requests.get(grid_endpoint, headers=headers)
if response.status_code == 200:
data = response.json()
# Extract the relevant forecast information
periods = data["properties"]["periods"]
# Process the forecast data into a simpler format
forecast = {
"location": data["properties"].get("relativeLocation", {}).get("properties", {}),
"updated": data["properties"].get("updated", ""),
"periods": []
}
for period in periods:
forecast["periods"].append({
"name": period.get("name", ""),
"temperature": period.get("temperature", ""),
"temperatureUnit": period.get("temperatureUnit", ""),
"windSpeed": period.get("windSpeed", ""),
"windDirection": period.get("windDirection", ""),
"shortForecast": period.get("shortForecast", ""),
"detailedForecast": period.get("detailedForecast", "")
})
return forecast
else:
print(f"Error getting forecast: {response.status_code} - {response.text}")
return {"error": f"API Error: {response.status_code}"}
except Exception as e:
print(f"Error in get_forecast: {e}")
return {"error": f"Exception: {str(e)}"}
# Example usage
# def do_weather():
# city = input("Enter city: ")
# state = input("Enter state: ")
# country = input("Enter country (default USA): ") or "USA"
# print(f"Getting weather for {city}, {state}, {country}...")
# weather_data = WeatherForecast(city, state, country)
# if "error" in weather_data:
# print(f"Error: {weather_data['error']}")
# else:
# print("\nWeather Forecast:")
# print(f"Location: {weather_data.get('location', {}).get('city', city)}, {weather_data.get('location', {}).get('state', state)}")
# print(f"Last Updated: {weather_data.get('updated', 'N/A')}")
# print("\nForecast Periods:")
# for period in weather_data.get("periods", []):
# print(f"\n{period['name']}:")
# print(f" Temperature: {period['temperature']}{period['temperatureUnit']}")
# print(f" Wind: {period['windSpeed']} {period['windDirection']}")
# print(f" Forecast: {period['shortForecast']}")
# print(f" Details: {period['detailedForecast']}")
# %%
def TickerValue(ticker_symbols):
api_key = os.getenv("TWELVEDATA_API_KEY", "")
if not api_key:
return {"error": f"Error fetching data: No API key for TwelveData"}
results = []
for ticker_symbol in ticker_symbols.split(','):
ticker_symbol = ticker_symbol.strip()
if ticker_symbol == "":
continue
url = f"https://api.twelvedata.com/price?symbol={ticker_symbol}&apikey={api_key}"
response = requests.get(url)
data = response.json()
if "price" in data:
logging.info(f"TwelveData: {ticker_symbol} {data}")
results.append({
"symbol": ticker_symbol,
"price": float(data["price"])
})
else:
logging.error(f"TwelveData: {data}")
results.append({
"symbol": ticker_symbol,
"price": "Unavailable"
})
return results[0] if len(results) == 1 else results
# Stock related function
def yfTickerValue(ticker_symbols):
"""
Look up the current price of a stock using its ticker symbol.
Args:
ticker_symbol (str): The stock ticker symbol (e.g., 'AAPL' for Apple)
Returns:
dict: Current stock information including price
"""
results = []
for ticker_symbol in ticker_symbols.split(','):
ticker_symbol = ticker_symbol.strip()
if ticker_symbol == "":
continue
# Create a Ticker object
try:
logging.info(f"Looking up {ticker_symbol}")
ticker = yf.Ticker(ticker_symbol)
# Get the latest market data
ticker_data = ticker.history(period="1d")
if ticker_data.empty:
results.append({"error": f"No data found for ticker {ticker_symbol}"})
continue
# Get the latest closing price
latest_price = ticker_data['Close'].iloc[-1]
# Get some additional info
results.append({ 'symbol': ticker_symbol, 'price': latest_price })
except Exception as e:
import traceback
logging.error(f"Error fetching data for {ticker_symbol}: {e}")
logging.error(traceback.format_exc())
results.append({"error": f"Error fetching data for {ticker_symbol}: {str(e)}"})
return results[0] if len(results) == 1 else results
# %%
def DateTime(timezone="America/Los_Angeles"):
"""
Returns the current date and time in the specified timezone in ISO 8601 format.
Args:
timezone (str): Timezone name (e.g., "UTC", "America/New_York", "Europe/London")
Default is "America/Los_Angeles"
Returns:
str: Current date and time with timezone in the format YYYY-MM-DDTHH:MM:SS+HH:MM
"""
try:
if timezone == 'system' or timezone == '' or not timezone:
timezone = 'America/Los_Angeles'
# Get current UTC time (timezone-aware)
local_tz = pytz.timezone("America/Los_Angeles")
local_now = datetime.now(tz=local_tz)
# Convert to target timezone
target_tz = pytz.timezone(timezone)
target_time = local_now.astimezone(target_tz)
return target_time.isoformat()
except Exception as e:
return {'error': f"Invalid timezone {timezone}: {str(e)}"}
async def AnalyzeSite(llm, model: str, url : str, question : str):
"""
Fetches content from a URL, extracts the text, and uses Ollama to summarize it.
Args:
url (str): The URL of the website to summarize
Returns:
str: A summary of the website content
"""
try:
# Fetch the webpage
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
}
logging.info(f"Fetching {url}")
response = requests.get(url, headers=headers, timeout=10)
response.raise_for_status()
logging.info(f"{url} returned. Processing...")
# Parse the HTML
soup = BeautifulSoup(response.text, "html.parser")
# Remove script and style elements
for script in soup(["script", "style"]):
script.extract()
# Get text content
text = soup.get_text(separator=" ", strip=True)
# Clean up text (remove extra whitespace)
lines = (line.strip() for line in text.splitlines())
chunks = (phrase.strip() for line in lines for phrase in line.split(" "))
text = " ".join(chunk for chunk in chunks if chunk)
# Limit text length if needed (Ollama may have token limits)
max_chars = 100000
if len(text) > max_chars:
text = text[:max_chars] + "..."
# Create Ollama client
# logging.info(f"Requesting summary of: {text}")
# Generate summary using Ollama
prompt = f"CONTENTS:\n\n{text}\n\n{question}"
response = llm.generate(model=model,
system="You are given the contents of {url}. Answer the question about the contents",
prompt=prompt)
#logging.info(response["response"])
return {
"source": "summarizer-llm",
"content": response["response"],
"metadata": DateTime()
}
except requests.exceptions.RequestException as e:
logging.error(f"Error fetching the URL: {e}")
return f"Error fetching the URL: {str(e)}"
except Exception as e:
logging.error(f"Error processing the website content: {e}")
return f"Error processing the website content: {str(e)}"
# %%
tools = [ {
"type": "function",
"function": {
"name": "TickerValue",
"description": "Get the current stock price of one or more ticker symbols. Returns an array of objects with 'symbol' and 'price' fields. Call this whenever you need to know the latest value of stock ticker symbols, for example when a user asks 'How much is Intel trading at?' or 'What are the prices of AAPL and MSFT?'",
"parameters": {
"type": "object",
"properties": {
"ticker": {
"type": "string",
"description": "The company stock ticker symbol. For multiple tickers, provide a comma-separated list (e.g., 'AAPL,MSFT,GOOGL').",
},
},
"required": ["ticker"],
"additionalProperties": False
}
}
}, {
"type": "function",
"function": {
"name": "AnalyzeSite",
"description": "Downloads the requested site and asks a second LLM agent to answer the question based on the site content. For example if the user says 'What are the top headlines on cnn.com?' you would use AnalyzeSite to get the answer. Only use this if the user asks about a specific site or company.",
"parameters": {
"type": "object",
"properties": {
"url": {
"type": "string",
"description": "The website URL to download and process",
},
"question": {
"type": "string",
"description": "The question to ask the second LLM about the content",
},
},
"required": ["url", "question"],
"additionalProperties": False
},
"returns": {
"type": "object",
"properties": {
"source": {
"type": "string",
"description": "Identifier for the source LLM"
},
"content": {
"type": "string",
"description": "The complete response from the second LLM"
},
"metadata": {
"type": "object",
"description": "Additional information about the response"
}
}
}
}
}, {
"type": "function",
"function": {
"name": "DateTime",
"description": "Get the current date and time in a specified timezone. For example if a user asks 'What time is it in Poland?' you would pass the Warsaw timezone to DateTime.",
"parameters": {
"type": "object",
"properties": {
"timezone": {
"type": "string",
"description": "Timezone name (e.g., 'UTC', 'America/New_York', 'Europe/London', 'America/Los_Angeles'). Default is 'America/Los_Angeles'."
}
},
"required": []
}
}
}, {
"type": "function",
"function": {
"name": "WeatherForecast",
"description": "Get the full weather forecast as structured data for a given CITY and STATE location in the United States. For example, if the user asks 'What is the weather in Portland?' or 'What is the forecast for tomorrow?' use the provided data to answer the question.",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "City to find the weather forecast (e.g., 'Portland', 'Seattle').",
"minLength": 2
},
"state": {
"type": "string",
"description": "State to find the weather forecast (e.g., 'OR', 'WA').",
"minLength": 2
}
},
"required": [ "city", "state" ],
"additionalProperties": False
}
}
}]
def llm_tools(tools):
return [tool for tool in tools if tool.get("enabled", False) == True]
def enabled_tools(tools: list[dict[str, Any]]) -> list[dict[str, Any]]:
return [{**tool, "enabled": True} for tool in tools]
tool_functions = [ 'DateTime', 'WeatherForecast', 'TickerValue', 'AnalyzeSite' ]
__all__ = [ 'tools', 'llm_tools', 'enabled_tools', 'tool_functions' ]
#__all__.extend(__tool_functions__) # type: ignore