Skip to content

refactor: streamline RAG query execution in Hivemind agent by replaci… #49

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
amindadgar merged 2 commits into from
Oct 1, 2025
Merged

refactor: streamline RAG query execution in Hivemind agent by replaci… #49

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
de0eef5
refactor: streamline RAG query execution in Hivemind agent by replaci…
amindadgar Sep 30, 2025
bb52bd3
fix: removed using codeClimate!
amindadgar Sep 30, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .github/workflows/production.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,6 @@ on:

jobs:
ci:
uses: TogetherCrew/operations/.github/workflows/ci.yml@main
uses: TogetherCrew/operations/.github/workflows/ci2.yml@main
secrets:
CC_TEST_REPORTER_ID: ${{ secrets.CC_TEST_REPORTER_ID }}
2 changes: 1 addition & 1 deletion .github/workflows/start.staging.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,6 @@ on: pull_request

jobs:
ci:
uses: TogetherCrew/operations/.github/workflows/ci.yml@main
uses: TogetherCrew/operations/.github/workflows/ci2.yml@main
secrets:
CC_TEST_REPORTER_ID: ${{ secrets.CC_TEST_REPORTER_ID }}
4 changes: 1 addition & 3 deletions requirements.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,4 @@ tc-temporal-backend==1.1.4
transformers[torch]==4.49.0
nest-asyncio==1.6.0
openai==1.93.0
tc-hivemind-backend==1.4.3
langchain==0.3.26
langchain-openai==0.3.27
tc-hivemind-backend==1.4.3
44 changes: 14 additions & 30 deletions tasks/hivemind/agent.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,18 +1,16 @@
import logging
import asyncio
from crewai import Agent, Crew, Task
from crewai.crews.crew_output import CrewOutput
from crewai.flow.flow import Flow, listen, start, router
from crewai.llm import LLM
from tasks.hivemind.classify_question import ClassifyQuestion
from tasks.hivemind.query_data_sources import make_rag_tool
from tasks.hivemind.query_data_sources import QueryDataSources
from pydantic import BaseModel
from crewai.tools import tool
from openai import OpenAI
from typing import Optional
from tasks.mongo_persistence import MongoPersistence
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.agents import AgentExecutor, create_openai_functions_agent


class AgenticFlowState(BaseModel):
Expand Down Expand Up @@ -156,35 +154,21 @@ def detect_question_type(self) -> str:

@router("rag")
def do_rag_query(self) -> str:
llm = ChatOpenAI(model="gpt-4o-mini-2024-07-18")
rag_tool = make_rag_tool(self.enable_answer_skipping, self.community_id, self.workflow_id)
tools = [rag_tool]

SYSTEM_INSTRUCTIONS = """\
You are a helpful assistant.
"""

prompt = ChatPromptTemplate.from_messages(
[
("system", SYSTEM_INSTRUCTIONS),
MessagesPlaceholder("chat_history", optional=True),
("human", "{input}"),
MessagesPlaceholder("agent_scratchpad"),
]
query_data_sources = QueryDataSources(
community_id=self.community_id,
enable_answer_skipping=self.enable_answer_skipping,
workflow_id=self.workflow_id,
)
agent = create_openai_functions_agent(llm, tools, prompt)

# Run the agent
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
return_intermediate_steps=False,
max_iterations=3,
)
try:
answer = asyncio.run(query_data_sources.query(self.state.user_query))
if answer is None:
answer = "NONE"
except Exception as e:
logging.error(f"RAG query execution failed: {e}")
answer = "NONE"

result = agent_executor.invoke({"input": self.state.user_query})
self.state.last_answer = result["output"]
self.state.last_answer = answer
self.state.retry_count += 1

return "stop"
Comment on lines +157 to 174
Copy link

@coderabbitai coderabbitai bot Sep 30, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Type inconsistency: last_answer expects CrewOutput | None but receives str.

Line 171 assigns a string value (answer) to self.state.last_answer, but the field is typed as CrewOutput | None (line 19). This mismatch will cause type checking failures and potential runtime issues downstream where CrewOutput is expected.

Apply this diff to fix the type annotation:

 class AgenticFlowState(BaseModel):
     user_query: str = ""
     retry_count: int = 0
-    last_answer: CrewOutput | None = None
+    last_answer: CrewOutput | str | None = None
     state: str = "continue"
     chat_history: str | None = None

Alternatively, if last_answer should remain CrewOutput | None, wrap the string response in a CrewOutput object before assignment.

📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
query_data_sources = QueryDataSources(
community_id=self.community_id,
enable_answer_skipping=self.enable_answer_skipping,
workflow_id=self.workflow_id,
)
agent = create_openai_functions_agent(llm, tools, prompt)
# Run the agent
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
return_intermediate_steps=False,
max_iterations=3,
)
try:
answer = asyncio.run(query_data_sources.query(self.state.user_query))
if answer is None:
answer = "NONE"
except Exception as e:
logging.error(f"RAG query execution failed: {e}")
answer = "NONE"
result = agent_executor.invoke({"input": self.state.user_query})
self.state.last_answer = result["output"]
self.state.last_answer = answer
self.state.retry_count += 1
return "stop"
class AgenticFlowState(BaseModel):
user_query: str = ""
retry_count: int = 0
last_answer: CrewOutput | str | None = None
state: str = "continue"
chat_history: str | None = None
🧰 Tools
🪛 Ruff (0.13.1)

167-167: Do not catch blind exception: Exception

(BLE001)

168-168: Use logging.exception instead of logging.error

Replace with exception

(TRY400)

🤖 Prompt for AI Agents 
In tasks/hivemind/agent.py around lines 157 to 174, last_answer is being set to
a plain string but its type is declared as CrewOutput | None (line 19); update
the code so the assigned value matches the declared type by either (A) wrapping
the string into a CrewOutput instance before assigning to self.state.last_answer
(create a CrewOutput with the appropriate fields populated from the RAG response
and use that), or (B) if the design intent is to allow plain strings, change the
state type annotation for last_answer to str | CrewOutput | None (and update any
downstream usages to handle the string case). Ensure you choose one approach and
make corresponding downstream type/usage updates so type checking passes.

Expand Down
11 changes: 9 additions & 2 deletions tasks/hivemind/query_data_sources.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,6 @@
import nest_asyncio
from dotenv import load_dotenv
from typing import Optional, Callable
from langchain.tools import tool
from tc_temporal_backend.client import TemporalClient
from tc_temporal_backend.schema.hivemind import HivemindQueryPayload
from temporalio.common import RetryPolicy
Expand Down Expand Up @@ -96,7 +95,15 @@ def make_rag_tool(enable_answer_skipping: bool, community_id: str, workflow_id:
Returns:
Callable: The RAG pipeline tool.
"""
@tool(return_direct=True)
try:
from langchain.tools import tool as lc_tool # type: ignore
except Exception:
# Fallback no-op decorator if LangChain is not installed/required
def lc_tool(*_args, **_kwargs):
def decorator(func):
return func
return decorator
@lc_tool(return_direct=True)
def get_rag_answer(query: str) -> str:
"""
Get the answer from the RAG pipeline
Expand Down