> ## Documentation Index
> Fetch the complete documentation index at: https://docs.orq.ai/llms.txt
> Use this file to discover all available pages before exploring further.
# Build a multi-agent HR system
> Build a multi-agent HR system with Python. Create specialized agents for benefits, PTO, and policy questions using memory and knowledge.
## TL;DR
* Build HR management agents using [Orq.ai](http://Orq.ai) Python SDK
* Connect tools (Python custom functions)
* Enable memory for context-aware conversations
* Add Knowledge Base with company policy documents
* Create multi-agent teams with specialized roles (benefits, PTO calculator)
## What is Agents API?
The [Orq.ai](http://Orq.ai) Agents API is a powerful framework within the [Orq.ai](http://Orq.ai) ecosystem that enables developers to build intelligent, autonomous AI agents capable of reasoning, using tools, maintaining context, and collaborating with other agents. It sits at the core of [Orq.ai](http://Orq.ai)'s agentic architecture, working seamlessly with other platform components:
* **Deployments**: Version-controlled configurations for your AI applications
* **Gateway**: Unified entry point for routing requests and managing LLM providers
* **Agents**: Autonomous entities that execute tasks using reasoning, tools, and context
## Common Problems Agents Solve
1. **Complex Task Orchestration:**
Traditional chatbots struggle with multi-step workflows. Agents can break down complex requests into subtasks and use an orchestrator to delegate them between sub-agents that communicate via the **A2A Protocol.**
2. **Persistent Memory & Personalization**
Unlike stateless API calls, Agents maintain conversation history and user preferences across sessions through Memory Stores, enabling truly personalized employee experiences.
3. **Grounding & Accuracy**
By integrating Knowledge Bases, Agents ground their responses in your company's actual policies and documents, reducing hallucinations
## **What we are going to build?**
You will build an HR management system using [Orq.ai](http://Orq.ai) Agents, where employees can get instant answers to HR questions, submit requests, and receive personalized support. You can follow along the steps in the [Google Colab notebook](https://colab.research.google.com/drive/1wL0wlcgZBggRmLhNNfl_rsCaKHtXhowv?usp=sharing). The system will include specialized agents for different HR domains (benefits questions, PTO calculation) orchestrated by a coordinator agent that intelligently routes requests. You'll learn to implement persistent memory for personalized interactions and knowledge base integration for policy-compliant answers.
```
graph LR
A[Employee Query] --> B{Check memory_stores}
B -->|Has Context| C[Query Knowledge Base]
B -->|No Context| C
C -->|Benefits Question| D[Benefits Agent]
C -->|Payroll Question| E[Payroll Agent]
D --> F[Generate Response]
E --> F
F --> H[Save to Memory]
H --> A
style C fill:#f0e1ff
style F fill:#e1ffe1
```
## Prerequisites
1. Python 3.8+ python or higher
2. An [**API Key**](https://docs.orq.ai/docs/administer/api-keys) ready to be used with the API.
3. [Orq.ai](http://Orq.ai) SDK installed
```
pip install orq-ai-sdk
```
4. A workspace with a [**Project**](https://docs.orq.ai/docs/projects/overview) named `agents`and Folder.
5. Copy the `path` property ( You will use it in code as `path="agent"` )
Agents are defined via JSON payloads or the Python SDK. Key elements include **Metadata** - to set the tone of the agent and **Configuration parameters** to fine tune the agent
**Agent Metadata**
* `role` : description and instructions
* `description` : short description of agent's purpose
* `instructions` : instructions for the agent behaviour
**Configuration parameters**
* `model`: choose from providers e.g. `openai/gpt-4o`, `anthropic/claude-3-5-sonnet` and set the number of `retries` on specific error codes
* `settings` : Configuration settings for the agent's behavior, such as:
* `max_iterations` : Maximum iterations(llm calls) before the agent will stop
* `max_execution_time` : Maximum time (in seconds) for the agent thinking process.
* `tools` : extend agent capabilities by providing access to custom functionality, there are built-in tools such as `current_date` , `google_search` and `web_scraper` and custom tools (http, code, function)
Run the code below to create an agent:
```python theme={"theme":{"light":"github-light","dark":"github-dark"}}
from orq_ai_sdk import Orq
import os
import json
with Orq(
api_key=os.getenv("ORQ_API_KEY", ""),
) as orq:
res = orq.agents.create(
key="policy_agent",
role="HR Policy Assistant",
description="Answers questions about company policies and procedures",
instructions="Provide clear, concise answers to HR policy questions. Always cite relevant policy sections.",
path="agent",
model={
"id": "openai/gpt-4o",
"retry": {
"count": 3,
"on_codes": [429, 500, 502, 503, 504],
},
},
settings={
"tools": [],
},
)
assert res is not None
# Formatted output
print("=" * 60)
print("AGENT CREATED SUCCESSFULLY")
print("=" * 60)
print(f"Agent ID: {res.id}")
print(f"Agent Key: {res.key}")
print(f"Display Name: {res.display_name}")
print(f"Status: {res.status}")
print(f"Role: {res.role}")
print(f"Description: {res.description}")
print(f"Path: {res.path}")
print(f"Project ID: {res.project_id}")
print("-" * 60)
```
Expected output:
```
============================================================
AGENT CREATED SUCCESSFULLY
============================================================
Agent ID: 01KB56Q4ZJH58RAEA7ZTSNHF4E
Agent Key: policy_agent
Display Name: policy_agent
Status: live
Role: HR Policy Assistant
Description: Answers questions about company policies and procedures
Path: agent
Project ID: 019aca1c-fe79-7000-937c-f907088aeaa8
------------------------------------------------------------
```
Next, you need to invoke a response from a newly created agent:
```python theme={"theme":{"light":"github-light","dark":"github-dark"}}
from orq_ai_sdk import Orq
import os
with Orq(
api_key=os.getenv("ORQ_API_KEY", ""),
) as orq:
res = orq.agents.responses.create(
agent_key="policy_agent_new",
message={
"role": "user",
"parts": [
{
"kind": "text",
"text": "What is the company policy on remote work and flexible hours?"
}
]
},
identity={
"id": "contact_john_doe_001",
"display_name": "John Doe",
"email": "john.doe@company.com",
"metadata": [
{
"department": "Engineering",
"role": "Software Developer",
}
],
"tags": [
"hr-inquiry",
"employee",
],
},
thread={
"id": "thread_hr_policy_001",
"tags": [
"hr-policy",
"remote-work",
],
},
background=False
)
assert res is not None
# Handle response
print("=" * 50)
print("Agent Response:")
print("=" * 50)
print(res.output[0].parts[0].text)
print("\n" + "=" * 50)
print("Usage Stats:")
print("=" * 50)
print(res.usage)
print("Task ID:", res.task_id)
```
Expected output
```
==================================================
Agent Response:
==================================================
While company policies can vary widely, a typical remote work and flexible hours policy might include the following components:
1. Eligibility: Employees eligible for remote work or flexible hours are often those whose responsibilities can effectively be carried out off-site or in non-standard hours. Specific roles or departments may be designated as eligible or ineligible. (Section Reference: Eligibility Criteria for Remote Work)
2. Approval Process: Employees usually need to seek approval from their manager to engage in remote work or flexible hours. Approval often depends on the nature of the job, employee performance, and team requirements. (Section Reference: Approval Procedures)
3. Work Schedule: Employees may set flexible hours as long as they fulfill their total weekly or monthly hours requirements. Core hours during which employees must be available might still be specified. (Section Reference: Work Schedule Requirements)
4. Availability and Communication: Employees working remotely should remain reachable during specified hours and use designated communication tools to ensure seamless collaboration. (Section Reference: Communication Expectations)
5. Performance Monitoring: Employers retain the right to assess employee performance using predefined metrics to ensure productivity is not affected by remote work arrangements. (Section Reference: Performance Monitoring)
6. Equipment and Security: The company might provide necessary tools and ensure employees have secure access to company data and networks. Employees are typically responsible for maintaining a distraction-free work environment. (Section Reference: Equipment and Data Security)
==================================================
Usage Stats:
==================================================
completion_tokens=312.0
prompt_tokens=43.0
total_tokens=355.0
prompt_tokens_details=CreateAgentResponseRequestPromptTokensDetails(cached_tokens=0, audio_tokens=0)
completion_tokens_details=CreateAgentResponseRequestCompletionTokensDetails(reasoning_tokens=0.0, accepted_prediction_tokens=0.0, rejected_prediction_tokens=0.0, audio_tokens=0)
==================================================
Task ID: 01KBF4VJWRPFNZZRK502AARHWE
```
After receiving a task response, you can continue the conversation by sending additional messages to the same agent. To enable multi-turn interactions where the agent maintains context from previous exchanges reference in the next run your previous task using `task_id`. To learn more see [Agents via the API](https://docs.orq.ai/docs/agents/run#continue-a-task)
```python theme={"theme":{"light":"github-light","dark":"github-dark"}}
from orq_ai_sdk import Orq
import os
with Orq(
api_key=os.getenv("ORQ_API_KEY", ""),
) as orq:
res = orq.agents.responses.create(
agent_key="policy_agent_new",
message={
"role": "user",
"parts": [
{
"kind": "text",
"text": "What is the company policy on remote work and flexible hours?"
}
]
}, task_id="01KBFNGE4W3N6A7RG9XNQ20212",
identity={
"id": "contact_john_doe_001",
"display_name": "John Doe",
"email": "john.doe@company.com",
"metadata": [
{
"department": "Engineering",
"role": "Software Developer",
}
],
"tags": [
"hr-inquiry",
"employee",
],
},
thread={
"id": "thread_hr_policy_001",
"tags": [
"hr-policy",
"remote-work",
],
},
background=False
)
assert res is not None
# Handle response
print("=" * 50)
print("Task ID:")
print("=" * 50)
print(res.task_id)
print("\n" + "=" * 50)
print("Agent Response:")
print("=" * 50)
print(res.output[0].parts[0].text)
print("\n" + "=" * 50)
```
Expected output:
```
==================================================
Task ID:
01KBFNGE4W3N6A7RG9XNQ20212
==================================================
Agent Response:
==================================================
Remote Work Policy: 2 days home office allowed
==================================================
```
To define custom business logic for HR operations and access add external systems, APIs, and custom functionality you can use Tools such as:
* **HTTP Tools:** Integrate with external APIs (e.g., weather, search, CRM).
* **Function Tools:** Inject custom logic via the OpenAI function-calling schema.
* **Code Tools:** Run small snippets of Python in secure sandboxes for data transformation.
To see a full reference of basic tools see [Tools with Agents](https://docs.orq.ai/docs/agents/build#add-tools)
First you need to define function calculating remaining PTO:
```python your_script.py theme={"theme":{"light":"github-light","dark":"github-dark"}}
from datetime import datetime, timedelta
def calculate_pto_remaining(start_date, accrual_rate, days_used, accrual_frequency):
"""Calculate remaining PTO days based on starting date and usage."""
# Parse start date if it's a string
if isinstance(start_date, str):
start_date = datetime.strptime(start_date, '%Y-%m-%d')
# Calculate days employed
today = datetime.now()
days_employed = (today - start_date).days
# Calculate accrual periods based on frequency
if accrual_frequency == 'monthly':
periods = days_employed / 30.44
elif accrual_frequency == 'biweekly':
periods = days_employed / 14
elif accrual_frequency == 'yearly':
periods = days_employed / 365.25
else: # daily
periods = days_employed
# Calculate total PTO accrued
pto_accrued = periods * accrual_rate
# Calculate remaining PTO
pto_remaining = pto_accrued - days_used
return {
"pto_remaining": round(pto_remaining, 2),
"pto_accrued": round(pto_accrued, 2),
"days_used": days_used,
"days_employed": days_employed,
"start_date": start_date.strftime('%Y-%m-%d'),
"accrual_rate": accrual_rate,
"accrual_frequency": accrual_frequency
}
# Define params for execution
params = {}
# Execute with params
result = calculate_pto_remaining(
params.get('start_date', '2024-01-01'),
params.get('accrual_rate', 1.25), # e.g., 1.25 days per month = 15 days/year
params.get('days_used', 5),
params.get('accrual_frequency', 'monthly')
)
```
**Tip: Converting Python Code to Payload String**
To convert your Python code into a JSON-safe string for the payload, use this shell command:
`cat your_script.py | jq -Rs '.' `
Add the function above as payload string to the Agent:
```python theme={"theme":{"light":"github-light","dark":"github-dark"}}
import os
from orq_ai_sdk import Orq
import orq_ai_sdk
from datetime import datetime
PTO_CALCULATOR_CODE = """from datetime import datetime, timedelta
def calculate_pto_remaining(start_date, accrual_rate, days_used, accrual_frequency):
\"\"\"Calculate remaining PTO days based on starting date and usage.\"\"\"
if isinstance(start_date, str):
start_date = datetime.strptime(start_date, '%Y-%m-%d')
today = datetime.now()
days_employed = (today - start_date).days
if accrual_frequency == 'monthly':
periods = days_employed / 30.44
elif accrual_frequency == 'biweekly':
periods = days_employed / 14
elif accrual_frequency == 'yearly':
periods = days_employed / 365.25
else:
periods = days_employed
pto_accrued = periods * accrual_rate
pto_remaining = pto_accrued - days_used
return {
"pto_remaining": round(pto_remaining, 2),
"pto_accrued": round(pto_accrued, 2),
"days_used": days_used,
"days_employed": days_employed,
"start_date": start_date.strftime('%Y-%m-%d'),
"accrual_rate": accrual_rate,
"accrual_frequency": accrual_frequency
}
result = calculate_pto_remaining(
start_date=start_date,
accrual_rate=accrual_rate,
days_used=days_used,
accrual_frequency=accrual_frequency
)
"""
print("=" * 60)
print("ORQ.AI HR POLICY AGENT WITH PTO CALCULATOR")
print("=" * 60)
print(f"Timestamp : {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print(f"Agent Key : pto_agent_final")
print(f"Model : openai/gpt-4o")
print(f"Tool Included : pto_calculator_final (code tool)")
print("=" * 60)
client = Orq(api_key="")
# TASK 1: Define PTO Calculator
print("\n[STEP 1] Creating PTO Calculator Tool...")
print("-" * 60)
tool_params_schema = {
"type": "object",
"properties": {
"start_date": {
"type": "string",
"format": "date",
"description": "Employee start date (YYYY-MM-DD)",
"default": "2024-01-01"
},
"accrual_rate": {
"type": "number",
"description": "PTO days accrued per period (e.g., 1.5 for monthly)",
"default": 1.25
},
"days_used": {
"type": "number",
"description": "Total PTO days already used",
"default": 0
},
"accrual_frequency": {
"type": "string",
"enum": ["daily", "biweekly", "monthly", "yearly"],
"description": "How PTO accrues",
"default": "monthly"
}
},
"required": ["start_date", "accrual_rate", "days_used", "accrual_frequency"],
"additionalProperties": False
}
# TASK 2: Create a PTO Calculator Tool
try:
tool = client.tools.create(
request=orq_ai_sdk.RequestBodyCodeExecutionTool(
key="pto_calculator_final",
type="code",
display_name="PTO Balance Calculator",
description="Calculates accrued and remaining PTO days based on hire date, accrual rules, and usage",
status="live",
requires_approval=False,
path="agent",
code_tool=orq_ai_sdk.RequestBodyCodeTool(
language="python",
code=PTO_CALCULATOR_CODE,
parameters=orq_ai_sdk.CreateToolRequestBodyParameters(**tool_params_schema),
)
)
)
print(f"✓ Tool created successfully: {tool.key}")
tool_created = True
except Exception as e:
error_str = str(e)
if "already exists" in error_str.lower() or "409" in error_str:
print(f"✓ Tool 'pto_calculator_final' already exists, continuing...")
tool_created = True
else:
print(f"⚠ Tool creation error: {error_str}")
print(" (Continuing anyway...)")
tool_created = False
if tool_created:
print("\n[STEP 2] Creating HR Policy Agent...")
print("-" * 60)
# TASK 3: Create an Agent using PTO tool
try:
agent = client.agents.create(
key="pto_agent_final",
role="HR Policy & Benefits Assistant",
description="Answers questions about company policies, benefits, and calculates PTO balances",
instructions="""You are an expert HR assistant. Answer policy questions clearly and cite sources when possible.
When a user asks about PTO balance or vacation days remaining, use the provided PTO calculator tool
to compute the accurate remaining balance based on hire date, accrual rate, and days already used.""",
model="openai/gpt-4o",
path="agent",
settings={
"max_iterations": 6,
"max_execution_time": 300,
"tools": [
{
"type": "code",
"key": "pto_calculator_final"
}
]
}
)
print(f"✓ Agent created successfully: {agent.key}")
agent_created = True
except Exception as e:
error_str = str(e)
if "already exists" in error_str.lower() or "409" in error_str:
print(f"✓ Agent 'pto_agent_final' already exists, continuing...")
agent_created = True
else:
print(f"⚠ Agent creation error: {error_str}")
agent_created = False
else:
print("\n[STEP 2] Skipping agent creation - tool not ready")
agent_created = False
# TASK 4: Invoke the agent
if agent_created:
print("\n[STEP 3] Sending Query to Agent...")
print("-" * 60)
print("User query: 'I started on 2024-03-15, I accrue 1.5 days per month,")
print(" I've used 8 days so far. How much PTO do I have left?'")
print("-" * 60)
try:
response = client.agents.responses.create(
agent_key="pto_agent_final",
message={
"role": "user",
"parts": [
{
"kind": "text",
"text": "I started on 2024-03-15, I accrue 1.5 days per month, I've used 8 days so far. How much PTO do I have left?"
}
]
}
)
print("\n" + "=" * 60)
print("AGENT RESPONSE")
print("=" * 60)
print(f"Task ID: {response.task_id}")
if hasattr(response, 'model'):
print(f"Model: {response.model}")
print("-" * 60)
if response.output:
for message in response.output:
if message.role == "agent":
for part in message.parts:
if part.kind == "text":
print(part.text)
print("-" * 60)
if hasattr(response, 'usage'):
print("Usage:", response.usage)
print("=" * 60)
except Exception as e:
print(f"\n⚠ Query error: {str(e)}")
else:
print("\n[STEP 3] Skipping query - agent not ready")
```
Expected output:
```
============================================================
ORQ.AI HR POLICY AGENT WITH PTO CALCULATOR
============================================================
Timestamp : 2025-12-02 15:28:28
Agent Key : pto_agent_final
Model : openai/gpt-4o
Tool Included : pto_calculator_final (code tool)
============================================================
[STEP 1] Creating PTO Calculator Tool...
------------------------------------------------------------
✓ Tool created successfully: pto_calculator_final
[STEP 2] Creating HR Policy Agent...
------------------------------------------------------------
✓ Agent created successfully: pto_agent_final
[STEP 3] Sending Query to Agent...
------------------------------------------------------------
User query: 'I started on 2024-03-15, I accrue 1.5 days per month,
I've used 8 days so far. How much PTO do I have left?'
============================================================
AGENT RESPONSE
============================================================
Task ID: 01KBFTVW8SQ4G0PD1RX2NP9V28
Model: openai/gpt-4o
------------------------------------------------------------
You have 4 days remaining.
------------------------------------------------------------
Usage: completion_tokens=41.0 prompt_tokens=415.0 total_tokens=456.0 prompt_tokens_details=CreateAgentResponseRequestPromptTokensDetails(cached_tokens=0, audio_tokens=0) completion_tokens_details=CreateAgentResponseRequestCompletionTokensDetails(reasoning_tokens=0.0, accepted_prediction_tokens=0.0, rejected_prediction_tokens=0.0, audio_tokens=0)
============================================================
```
To remember employee preferences and store conversation history we will add in the next step the `memory_stores`, that would help persist context across sessions.
1. **Create a** **memory store**
```python theme={"theme":{"light":"github-light","dark":"github-dark"}}
import os
from orq_ai_sdk import Orq
orq = Orq(
api_key=os.getenv("ORQ_API_KEY", ""),
)
try:
memory_store = orq.memory_stores.create(
request={
"key": "hr_employee_pto_data",
"description": "Store for employee PTO details, hire dates, and accrual rates",
"path": "agents",
"embedding_config": {
"model": "openai/text-embedding-3-small"
}
}
)
print(f"✓ Memory store created: {memory_store.key}")
except Exception as e:
if "already exists" in str(e).lower():
print("✓ Memory store already exists, continuing...")
else:
raise e
```
2. **Run HR Agent with memory store**
```python theme={"theme":{"light":"github-light","dark":"github-dark"}}
from orq_ai_sdk import Orq
import os
from datetime import datetime
import os
from orq_ai_sdk import Orq
orq = Orq(
api_key=os.getenv("ORQ_API_KEY", ""),
)
# ------------------------------------------------------------------
# Configuration & Constants
# ------------------------------------------------------------------
print("-" * 70)
# Works
# Create agent with BRAND NEW KEY and CORRECT schema
try:
key = 'hr_pto_mem_v4'
agent = orq.agents.create(
key=key,
role="HR Policy & Benefits Assistant (with Memory)",
description="Remembers employee PTO details and answers policy questions",
instructions="""
You are a smart HR assistant with memory.
- Always check if the user has previously shared their hire date, accrual rate, or PTO usage using memory tools.
- If they have, recall and use that data instead of asking again.
- When new PTO details are shared, save them to memory for future conversations.
- Use the PTO calculator tool to give precise remaining balance.
""".strip(),
memory_stores=["hr_employee_pto_data"],
model="openai/gpt-4o",
path="agent",
settings={
"max_iterations": 8,
"max_execution_time": 300,
"tools": [
{
"key": "pto_calculator3",
"type": "code",
"display_name": "PTO Balance Calculator",
"description": "Calculates accrued and remaining PTO days",
},
{"type": "retrieve_memory_stores"},
{"type": "query_memory_store"},
{"type": "write_memory_store"},
{"type": "delete_memory_document"}
]
}
)
print(f"✓ Agent created: {agent.id}")
print(f"✓ Key: {key}")
print("-" * 70)
# Create a response
response = orq.agents.responses.create(
agent_key=key,
memory={"entity_id": f"user_{os.getenv('USER_ID', 'john_doe')}"},
message={
"role": "user",
"parts": [{
"kind": "text",
"text": "I started on 2024-03-15, I accrue 1.5 days per month, I've used 8 days so far. How much PTO do I have left?"
}]
}
)
print("✓ Agent execution completed!")
print("-" * 70)
print("FINAL RESPONSE:")
print("-" * 70)
print(f"Response ID: {response.id}")
print(f"Task ID: {response.task_id}")
print(f"Output: {response.output[0].parts[0].text if response.output else 'Processing...'}")
print("=" * 70)
except Exception as e:
print(e)
orq.agents.delete(agent_key='hr_pto_mem_v2')
print('deleted')
```
Expected output:
```
----------------------------------------------------------------------
✓ Agent created: 01KBFVZ97844Q62GMT6X6HP2K1
✓ Key: hr_pto_mem_v4
----------------------------------------------------------------------
✓ Agent execution completed!
----------------------------------------------------------------------
FINAL RESPONSE:
----------------------------------------------------------------------
Response ID: 01KBFVZ9B699RR32XNVSZJTRED
Task ID: 01KBFVZ9B8JE1GKRRCYVRNE1CE
Output: Based on your start date of March 15, 2024, an accrual rate of 1.5 days per month, and the fact that you've used 8 days so far, you have approximately 8.25 days of PTO remaining.
======================================================================
```
3. List memories in the memory store
**Memory Store** contains **Memories** (representing entities), which in turn contain documents (information). Here is an example look up of memories in a memory store:
```python theme={"theme":{"light":"github-light","dark":"github-dark"}}
from orq_ai_sdk import Orq
import os
orq = Orq(
api_key=os.getenv("ORQ_API_KEY", ""),
)
# List all memories
response = orq.memory_stores.list_memories(
memory_store_key="hr_employee_pto_data",
limit=50
)
print(f"Total memories: {len(response.data)}\n")
for memory in response.data:
print(f"Entity: {memory.entity_id}")
print(f"Content: {memory.content}\n")
```
Expected output
```
Total memories: 1
Entity: user_john_doe
Content: Employee started on 2024-03-15 with an accrual rate of 1.5 PTO days per month. Has used 8 days so far.
```
To learn about other use cases of memory stores see [Using Memory Stores](https://docs.orq.ai/docs/memory-stores/using-memory-stores)
Make sure that your Agent grounds its responses based on your company policies. To do that we will add a Knowledge Base for contextual accuracy.
1. Create a Knowledge Base
```python theme={"theme":{"light":"github-light","dark":"github-dark"}}
from orq_ai_sdk import Orq
import os
orq = Orq(
api_key=os.getenv("ORQ_API_KEY", ""),
)
try:
res = orq.knowledge.create(
request={
"type": "internal",
"key": "agents",
"path": "agents",
"embedding_model": "text-embedding-3-large",
"top_k": 5,
"threshold": 0.7,
"search_type": "hybrid_search",
"is_private_model": False,
}
)
print("Knowledge base created:", res)
except Exception as e:
if "already exists" in str(e).lower():
print('Knowledge base already exists, fetching...')
res = orq.knowledge.get(request={"key": "agents"})
print("Existing knowledge base:", res)
else:
raise e
```
2. Upload a file
[Orq.ai](http://Orq.ai) supports document of the following format: TXT, PDF, DOCX, CSV, XML
Make sure to change the file path
```python theme={"theme":{"light":"github-light","dark":"github-dark"}}
import os
from orq_ai_sdk import Orq
orq = Orq(
api_key=os.getenv("ORQ_API_KEY", ""),
)
# 1. Upload the file (purpose="retrieval" is required for RAG/knowledge use)
try:
file_response = orq.files.create( # Changed from client to orq
file={
"file_name": "policy.pdf",
"content": open("path/to/your/policy.pdf", "rb"), # Update to your file path
},
purpose="retrieval"
)
print(f"File uploaded successfully → ID: {file_response.id}")
except FileNotFoundError:
print("Error: File not found at the specified path. Please check the file path and try again.")
except PermissionError:
print("Error: Permission denied. Make sure you have read access to the file.")
except Exception as e:
print(f"Upload failed: {str(e)}")
```
3. Create a datasource
A **Datasource** is the integral part of the Knowledge Base, it holds chunks of data within which a model can search and make retrievals returned within a RAG use case. A Knowledge base can hold any number of Datasources.
```python theme={"theme":{"light":"github-light","dark":"github-dark"}}
from orq_ai_sdk import Orq
import os
orq = Orq(
api_key=os.getenv("ORQ_API_KEY", ""),
)
file_id = file_response.id
knowledge_id = res.id
# Changed from client to orq
data_source = orq.knowledge.create_datasource(
knowledge_id=knowledge_id,
file_id=file_id
)
print(f"Data source created → ID: {data_source.id}")
```
4. Integrate Knowledge Base into the Agent
```python theme={"theme":{"light":"github-light","dark":"github-dark"}}
import os
from orq_ai_sdk import Orq
client = Orq(api_key=os.getenv("ORQ_API_KEY"))
agent = client.agents.create(
key="hr_policy_agent",
role="HR Policy Assistant",
description="Answers questions about company policies by searching uploaded policy documents",
instructions="""
You are an expert HR Policy Assistant with access to company policy documents.
CRITICAL INSTRUCTIONS:
1. ALWAYS use the query_knowledge_base tool FIRST to search for relevant policies
2. Base your answers on the retrieved policy documents
3. Cite specific policy sections when available (e.g., "According to Section 3.2...")
4. If no relevant information is found in the knowledge base, clearly state this
5. Be clear, concise, and professional in your responses
""".strip(),
model="openai/gpt-4o",
path="agent",
knowledge_bases=["agents"], # Link to the KB created in Step 1
settings={
"max_iterations": 8,
"max_execution_time": 300,
"tools": [
{"type": "query_knowledge_base"},
{"type": "current_date"}
]
}
)
print(f"✓ Agent created successfully!")
print(f" Agent ID: {agent.id}")
print(f" Agent Key: {agent.key}")
print(f" Knowledge Base: agents")
print(f" Tools enabled: query_knowledge_base, current_date")
```
Expected output
```
Agent created successfully!
Agent ID: 01KBH123XYZ456ABC789
Agent Key: hr_policy_agent
Knowledge Base: agents
Tools enabled: query_knowledge_base, current_date
```
Next, we define a`team_of_agents` with specialized roles and task delegation. Above we already defined the main Agent, which is `hr-policy-agent`, which calculates PTO. Next, we add a sub-agent: `benefits_agent` that handles questions related to health benefits and retirement plan inquiries.
1. **Create sub-agent**:`benefits_agent`
```python theme={"theme":{"light":"github-light","dark":"github-dark"}}
import os
from orq_ai_sdk import Orq
client = Orq(
api_key=os.getenv("ORQ_API_KEY"),
)
# Simple Benefits Specialist — no knowledge base
benefits_agent = client.agents.create(
request={
"key": "hr_benefits_specialist",
"role": "Benefits Enrollment Expert",
"description": "Helps employees with health insurance, 401k, PTO, and all benefits questions",
"instructions": "Answer clearly and kindly. Use current_date tool when needed for eligibility or open enrollment periods.",
"model": "openai/gpt-4o",
"path": "agents", # make sure "agents" project exists
"settings": {
"max_iterations": 5,
"max_execution_time": 300,
"tools": [
{"type": "current_date"} # very useful for benefits questions
]
}
}
)
print(f"Benefits agent created successfully!")
print(f"ID : {benefits_agent.id}")
print(f"Key: {benefits_agent.key}")
```
Expected output:
```
Benefits agent created successfully!
ID : 01KAGF9YQYZSYY1FK3RRB3VZHM
Key: hr_benefits_specialist
```
2. \*\*Create Orchestrator \*\*
We have created two agents with the following reference keys 2. `hr_benefits_specialist` : answers general HR questions (benefits, sick leave etc.) 2. `hr_policy_agent` : calculates PTO remaining based on starting date
We need to manage them via orchestrator that will delegate the tasks to one of the agents. To do that we need to define a`team_of_agents` array in your orchestrator agent configuration:
```
"team_of_agents": [
{
"key": "hr_benefits_specialist",
"role": "Benefits Enrollment Expert"
},
{
"key": "hr_policy_agent",
"role": "PTO & Vacation Balance Calculator"
}
```
Full code snippet creating an orchestrator with :
* **Required Tools**: Add `retrieve_agents` and `call_sub_agent` tools to your orchestrator’s configuration to enable sub-agent discovery and delegation
```python theme={"theme":{"light":"github-light","dark":"github-dark"}}
import os
from orq_ai_sdk import Orq
client = Orq(
api_key=os.getenv("ORQ_API_KEY"),
)
# === HR ORCHESTRATOR AGENT ===
orchestrator = client.agents.create(
request={
"key": "hr_orchestrator",
"role": "HR Coordinator Assistant",
"description": "A friendly HR orchestrator that helps employees by coordinating specialized HR sub-agents",
"path": "agents", # matches your project structure
"model": "openai/gpt-4o",
"instructions": """
You are a helpful and professional HR coordinator. Your job is to understand the employee's question and delegate it to the right specialist agent.
Available specialists:
- hr_benefits_specialist → for health insurance, 401k, open enrollment, benefits eligibility
- hr_policy_agent → for PTO balance, vacation days, accrual rates, "how much time do I have left?"
Workflow:
1. Use the retrieve_agents tool to confirm available agents.
2. Analyze the user’s question.
3. Call the correct sub-agent using call_sub_agent with a clear, direct instruction.
4. Wait for their response.
5. Return their answer EXACTLY as provided — do not rephrase or add commentary.
6. If unsure, default to hr_benefits_specialist.
Be efficient. Do not overthink. Delegate fast and accurately.
""".strip(),
"settings": {
"max_iterations": 15,
"max_execution_time": 600,
"tool_approval_required": "none",
"tools": [
{"type": "current_date"},
{"type": "retrieve_agents"},
{"type": "call_sub_agent"}
]
},
"team_of_agents": [
{
"key": "hr_benefits_specialist",
"role": "Benefits Enrollment Expert"
},
{
"key": "hr_policy_agent",
"role": "PTO & Vacation Balance Calculator"
}
],
"memory_stores": []
}
)
print("HR Orchestrator created successfully!")
print(f"Agent ID : {orchestrator.id}")
print(f"Agent Key: {orchestrator.key}")
```
Expected output:
```
HR Orchestrator created successfully!
Agent ID : 01KAGGT36ZK06ZB459JAF061XX
Agent Key: hr_orchestrator
```
Invoking the orchestrator
```python theme={"theme":{"light":"github-light","dark":"github-dark"}}
import os
from orq_ai_sdk import Orq
client = Orq(
api_key=os.getenv("ORQ_API_KEY"),
)
# Invoke the orchestrator (persistent agent)
task = client.agents.invoke(
key="hr_orchestrator",
message={
"role": "user",
"parts": [
{
"kind": "text",
"text": "I started on 2024-03-15, I accrue 1.5 days per month, and I've used 8 days so far. How much PTO do I have left?"
}
]
}
)
print(f"Orchestrator task started!")
print(f"Task ID: {task.id}")
```
Expected output:
```
Orchestrator task started!
Task ID: 01KAGH5C5RMFCR8T2RNFE6PK3D
```
## Troubleshooting & Best Practices
**Knowledge Base Not Returning Results**
* Confirm datasource creation completed successfully
* Verify `query_knowledge_base` tool is enabled
* Check `threshold` value (lower = more permissive, try 0.5 instead of 0.7)
* Ensure documents were successfully chunked and embedded
**Sub-Agent Delegation Issues**
* Verify both `retrieve_agents` and `call_sub_agent` tools are enabled
* Check `team_of_agents` array includes correct agent keys
* Ensure sub-agents exist and have matching keys
* Review orchestrator instructions for clarity on routing logic
**1. Error Handling & Resilience**
```python theme={"theme":{"light":"github-light","dark":"github-dark"}}
try:
response = client.agents.responses.create(
agent_key="hr_orchestrator",
memory={"entity_id": user_id},
message=message
)
except Exception as e:
logger.error(f"Agent execution failed: {str(e)}")
# Implement fallback logic or retry mechanism
```
**2. Rate Limiting & Timeouts**
* Set appropriate `max_execution_time` (300s for simple tasks, 600s for complex)
* Implement exponential backoff for retries
* Monitor API usage to stay within quotas
**3. Monitoring & Observability**
```python theme={"theme":{"light":"github-light","dark":"github-dark"}}
# Log agent interactions for debugging
logger.info(f"Agent: {response.id}, Tokens: {response.usage}, Duration: {response.latency_ms}ms")
```
* Track token usage and costs per agent/user
* Monitor average response times
* Set up alerts for failed executions
* Review agent conversation histories periodically
**Cost Optimization**
* Set `max_iterations` conservatively (5-8 for most use cases)
* Use cheaper models (e.g., `gpt-4o-mini`) for sub-agents handling simple tasks
* Monitor and optimize tool usage patterns
* Implement response caching for common queries
**Versioning & Updates**
* Use agent `key` versioning (e.g., `hr_orchestrator_v2`)
* Test changes in non-production environments first
* Maintain backward compatibility when updating tools
**Reduce Latency**
* Minimize `max_iterations` to prevent unnecessary reasoning loops
* Use streaming for real-time user feedback
* Pre-load frequently accessed knowledge base content
* Cache agent configurations where possible
**Improve Accuracy**
* Write clear, specific agent instructions with examples
* Use higher quality embedding models for knowledge bases (`text-embedding-3-large`)
* Increase `top_k` for knowledge base queries if retrieval seems incomplete
* Regularly review and refine agent instructions based on user feedback
**Token Efficiency**
* Keep instructions concise but clear
* Limit conversation history length in memory stores
* Use smaller models for simple sub-agents
* Implement smart context pruning for long conversations
## Conclusion
You've built a production-ready HR management system that showcases the power of [Orq.ai](http://Orq.ai)'s \*\*Multi-Agent architecture \*\*with `benefits_agent`and `hr-policy-agent`. Starting from a simple agent, you've progressed to a two-agent orchestration with persistent memory, knowledge base grounding, custom tools, and enterprise-grade observability through one unified API.
**To learn more explore the documentation:**