From 82657602e763f2becf6dd57676329f9f43171eeb Mon Sep 17 00:00:00 2001 From: Madhu Date: Mon, 27 Jan 2025 22:31:13 +0530 Subject: [PATCH 01/13] New Proj: tool use with r1 demo --- .../ai_r1-tooluse-langroid/README.md | 0 .../ai_r1-tooluse-langroid/main.py | 0 .../ai_r1-tooluse-langroid/requirements.txt | 0 .../ai_r1-tooluse-langroid/test.py | 147 ++++++++++++++++++ 4 files changed, 147 insertions(+) create mode 100644 ai_agent_tutorials/ai_r1-tooluse-langroid/README.md create mode 100644 ai_agent_tutorials/ai_r1-tooluse-langroid/main.py create mode 100644 ai_agent_tutorials/ai_r1-tooluse-langroid/requirements.txt create mode 100644 ai_agent_tutorials/ai_r1-tooluse-langroid/test.py diff --git a/ai_agent_tutorials/ai_r1-tooluse-langroid/README.md b/ai_agent_tutorials/ai_r1-tooluse-langroid/README.md new file mode 100644 index 0000000..e69de29 diff --git a/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py b/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py new file mode 100644 index 0000000..e69de29 diff --git a/ai_agent_tutorials/ai_r1-tooluse-langroid/requirements.txt b/ai_agent_tutorials/ai_r1-tooluse-langroid/requirements.txt new file mode 100644 index 0000000..e69de29 diff --git a/ai_agent_tutorials/ai_r1-tooluse-langroid/test.py b/ai_agent_tutorials/ai_r1-tooluse-langroid/test.py new file mode 100644 index 0000000..0350d0d --- /dev/null +++ b/ai_agent_tutorials/ai_r1-tooluse-langroid/test.py @@ -0,0 +1,147 @@ + +import os +from typing import List +import fire + +from langroid.pydantic_v1 import BaseModel, Field +import langroid as lr +from langroid.utils.configuration import settings +from langroid.agent.tool_message import ToolMessage +from langroid.agent.tools.orchestration import FinalResultTool +import langroid.language_models as lm +from rich.prompt import Prompt +from langroid.agent.chat_document import ChatDocument + +# for best results: +DEFAULT_LLM = lm.OpenAIChatModel.GPT4o + +os.environ["TOKENIZERS_PARALLELISM"] = "false" + +# (1) Define the desired structure via Pydantic. +# Here we define a nested structure for City information. +# The "Field" annotations are optional, and are included in the system message +# if provided, and help with generation accuracy. + + +class CityData(BaseModel): + population: int = Field(..., description="population of city") + country: str = Field(..., description="country of city") + + +class City(BaseModel): + name: str = Field(..., description="name of city") + details: CityData = Field(..., description="details of city") + + +# (2) Define the Tool class for the LLM to use, to produce the above structure. +class CityTool(lr.agent.ToolMessage): + """Present information about a city""" + + request: str = "city_tool" + purpose: str = """ + To present AFTER user gives a city name, + with all fields of the appropriate type filled out; + """ + city_info: City = Field(..., description="information about a city") + + def handle(self) -> FinalResultTool: + """Handle LLM's structured output if it matches City structure""" + print("SUCCESS! Got Valid City Info") + return FinalResultTool(answer=self.city_info) + + @staticmethod + def handle_message_fallback( + agent: lr.ChatAgent, msg: str | ChatDocument + ) -> str | None: + """ + We end up here when there was no recognized tool msg from the LLM; + In this case use the AgentDoneTool with content set to + the original message content. + """ + if isinstance(msg, ChatDocument) and msg.metadata.sender == lr.Entity.LLM: + return f""" + You forgot to use the TOOL/Function `{CityTool.name()}`. + Please use this tool to present the city info. + """ + + @classmethod + def examples(cls) -> List["ToolMessage"]: + # Used to provide few-shot examples in the system prompt + return [ + cls( + city_info=City( + name="San Francisco", + details=CityData( + population=800_000, + country="USA", + ), + ) + ) + ] + + +def app( + m: str = DEFAULT_LLM, # model + d: bool = False, # pass -d to enable debug mode (see prompts etc) + nc: bool = False, # pass -nc to disable cache-retrieval (i.e. get fresh answers) +): + settings.debug = d + settings.cache = not nc + # create LLM config + llm_cfg = lm.OpenAIGPTConfig( + chat_model=m or DEFAULT_LLM, + chat_context_length=32000, # set this based on model + max_output_tokens=1000, + temperature=0.2, + stream=True, + timeout=45, + ) + + # Recommended: First test if basic chat works with this llm setup as below: + # Once this works, then you can try the rest of the example. + # + # agent = lr.ChatAgent( + # lr.ChatAgentConfig( + # llm=llm_cfg, + # ) + # ) + # + # agent.llm_response("What is 3 + 4?") + # + # task = lr.Task(agent) + # verify you can interact with this in a chat loop on cmd line: + # task.run("Concisely answer some questions") + + # Define a ChatAgentConfig and ChatAgent + + config = lr.ChatAgentConfig( + llm=llm_cfg, + system_message=f""" + You will receive a city name, + and you must use the TOOL/FUNCTION `{CityTool.name()}` to generate/present + information about the city. + """, + ) + + agent = lr.ChatAgent(config) + + # (4) Enable the Tool for this agent --> this auto-inserts JSON instructions + # and few-shot examples (specified in the tool defn above) into the system message + agent.enable_message(CityTool) + + # (5) Create task specialized to return City object + task: City | None = lr.Task(agent, interactive=False)[City] + + while True: + city = Prompt.ask("Enter a city name") + if city in ["q", "x"]: + break + result: City | None = task.run(city) + if result: + print(f"City Info: {result}") + else: + print("No valid city info found.") + + +if __name__ == "__main__": + fire.Fire(app) \ No newline at end of file From 4a69ac7224408a5363cc9c59ac4510672d6112d6 Mon Sep 17 00:00:00 2001 From: Madhu Date: Tue, 28 Jan 2025 21:57:38 +0530 Subject: [PATCH 02/13] New Proj: tool use with r1 demo --- .../ai_r1-tooluse-langroid/main.py | 196 ++++++++++++++++++ .../ai_r1-tooluse-langroid/requirements.txt | 4 + 2 files changed, 200 insertions(+) diff --git a/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py b/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py index e69de29..0b6c599 100644 --- a/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py +++ b/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py @@ -0,0 +1,196 @@ +from typing import Optional, List, Dict, Any +import os +import time +import streamlit as st +from openai import OpenAI +import anthropic +from dotenv import load_dotenv +from rich import print as rprint +from rich.panel import Panel +from prompt_toolkit import PromptSession +from prompt_toolkit.styles import Style + +# Model Constants +DEEPSEEK_MODEL: str = "deepseek-reasoner" +CLAUDE_MODEL: str = "claude-3-5-sonnet-20241022" + +# Load environment variables +load_dotenv() + +class ModelChain: + """ + A class to handle interactions with DeepSeek and Claude models. + """ + def __init__(self, deepseek_api_key: str, anthropic_api_key: str) -> None: + """ + Initialize the ModelChain with API keys. + + Args: + deepseek_api_key: API key for DeepSeek + anthropic_api_key: API key for Anthropic + """ + self.deepseek_client = OpenAI( + api_key=deepseek_api_key, + base_url="https://api.deepseek.com" + ) + self.claude_client = anthropic.Anthropic(api_key=anthropic_api_key) + + self.deepseek_messages: List[Dict[str, str]] = [] + self.claude_messages: List[Dict[str, Any]] = [] + self.current_model: str = CLAUDE_MODEL + self.show_reasoning = True + + def set_model(self, model_name): + self.current_model = model_name + + def get_model_display_name(self): + return self.current_model + + def get_deepseek_reasoning(self, user_input: str) -> str: + """ + Get reasoning from DeepSeek model. + + Args: + user_input: User's input text + + Returns: + str: Reasoning content from DeepSeek + """ + start_time = time.time() + self.deepseek_messages.append({"role": "user", "content": user_input}) + + if self.show_reasoning: + rprint("\n[blue]Reasoning Process[/]") + + response = self.deepseek_client.chat.completions.create( + model=DEEPSEEK_MODEL, + max_tokens=1, + messages=self.deepseek_messages, + stream=True + ) + + reasoning_content = "" + final_content = "" + reasoning_placeholder = st.empty() + + for chunk in response: + if chunk.choices[0].delta.reasoning_content: + reasoning_piece = chunk.choices[0].delta.reasoning_content + reasoning_content += reasoning_piece + reasoning_placeholder.markdown(reasoning_content) + if self.show_reasoning: + print(reasoning_piece, end="", flush=True) + elif chunk.choices[0].delta.content: + final_content += chunk.choices[0].delta.content + + elapsed_time = time.time() - start_time + if elapsed_time >= 60: + time_str = f"{elapsed_time/60:.1f} minutes" + else: + time_str = f"{elapsed_time:.1f} seconds" + rprint(f"\n\n[yellow]Thought for {time_str}[/]") + + if self.show_reasoning: + print("\n") + return reasoning_content + + def get_claude_response(self, user_input: str, reasoning: str) -> str: + """ + Get response from Claude model. + + Args: + user_input: User's input text + reasoning: Reasoning from DeepSeek + + Returns: + str: Claude's response + """ + user_message = { + "role": "user", + "content": [{"type": "text", "text": user_input}] + } + + assistant_prefill = { + "role": "assistant", + "content": [{"type": "text", "text": f"{reasoning}"}] + } + + messages = [user_message, assistant_prefill] + response_placeholder = st.empty() + + rprint(f"[green]{self.get_model_display_name()}[/]", end="") + + try: + with self.claude_client.messages.stream( + model=self.current_model, + messages=messages, + max_tokens=8000 + ) as stream: + full_response = "" + for text in stream.text_stream: + full_response += text + response_placeholder.markdown(full_response) + + self.claude_messages.extend([user_message, { + "role": "assistant", + "content": [{"type": "text", "text": full_response}] + }]) + self.deepseek_messages.append({"role": "assistant", "content": full_response}) + + print("\n") + return full_response + + except Exception as e: + rprint(f"\n[red]Error in response: {str(e)}[/]") + return "Error occurred while getting response" + +def main() -> None: + """Main function to run the Streamlit app.""" + st.title("AI Assistant") + + # Sidebar for API keys + with st.sidebar: + st.header("API Configuration") + deepseek_api_key = st.text_input("DeepSeek API Key", type="password") + anthropic_api_key = st.text_input("Anthropic API Key", type="password") + show_reasoning = st.toggle("Show Reasoning Process", value=True) + + if st.button("Clear Chat History"): + st.session_state.messages = [] + st.experimental_rerun() + + # Initialize session state for messages + if "messages" not in st.session_state: + st.session_state.messages = [] + + # Display chat messages + for message in st.session_state.messages: + with st.chat_message(message["role"]): + st.markdown(message["content"]) + + # Chat input + if prompt := st.chat_input("What would you like to know?"): + if not deepseek_api_key or not anthropic_api_key: + st.error("Please enter both API keys in the sidebar.") + return + + # Initialize ModelChain + chain = ModelChain(deepseek_api_key, anthropic_api_key) + + # Add user message to chat + st.session_state.messages.append({"role": "user", "content": prompt}) + with st.chat_message("user"): + st.markdown(prompt) + + # Get AI response + with st.chat_message("assistant"): + if show_reasoning: + reasoning = chain.get_deepseek_reasoning(prompt) + else: + reasoning = "" + + response = chain.get_claude_response(prompt, reasoning) + st.session_state.messages.append({"role": "assistant", "content": response}) + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/ai_agent_tutorials/ai_r1-tooluse-langroid/requirements.txt b/ai_agent_tutorials/ai_r1-tooluse-langroid/requirements.txt index e69de29..5dc79a4 100644 --- a/ai_agent_tutorials/ai_r1-tooluse-langroid/requirements.txt +++ b/ai_agent_tutorials/ai_r1-tooluse-langroid/requirements.txt @@ -0,0 +1,4 @@ +streamlit +openai +anthropic +python-dotenv From b87c6539cb97cb177ded6c691d3052ba104f683f Mon Sep 17 00:00:00 2001 From: Madhu Date: Tue, 28 Jan 2025 22:24:55 +0530 Subject: [PATCH 03/13] feature 1 - requirmements.txt --- .../ai_r1-tooluse-langroid/main.py | 110 +++++++----------- 1 file changed, 43 insertions(+), 67 deletions(-) diff --git a/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py b/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py index 0b6c599..e94023f 100644 --- a/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py +++ b/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py @@ -18,17 +18,7 @@ CLAUDE_MODEL: str = "claude-3-5-sonnet-20241022" load_dotenv() class ModelChain: - """ - A class to handle interactions with DeepSeek and Claude models. - """ def __init__(self, deepseek_api_key: str, anthropic_api_key: str) -> None: - """ - Initialize the ModelChain with API keys. - - Args: - deepseek_api_key: API key for DeepSeek - anthropic_api_key: API key for Anthropic - """ self.deepseek_client = OpenAI( api_key=deepseek_api_key, base_url="https://api.deepseek.com" @@ -47,21 +37,9 @@ class ModelChain: return self.current_model def get_deepseek_reasoning(self, user_input: str) -> str: - """ - Get reasoning from DeepSeek model. - - Args: - user_input: User's input text - - Returns: - str: Reasoning content from DeepSeek - """ start_time = time.time() self.deepseek_messages.append({"role": "user", "content": user_input}) - if self.show_reasoning: - rprint("\n[blue]Reasoning Process[/]") - response = self.deepseek_client.chat.completions.create( model=DEEPSEEK_MODEL, max_tokens=1, @@ -71,27 +49,23 @@ class ModelChain: reasoning_content = "" final_content = "" - reasoning_placeholder = st.empty() - for chunk in response: - if chunk.choices[0].delta.reasoning_content: - reasoning_piece = chunk.choices[0].delta.reasoning_content - reasoning_content += reasoning_piece - reasoning_placeholder.markdown(reasoning_content) - if self.show_reasoning: - print(reasoning_piece, end="", flush=True) - elif chunk.choices[0].delta.content: - final_content += chunk.choices[0].delta.content + # Create expander for reasoning + with st.expander("💭 Reasoning Process", expanded=True): + reasoning_placeholder = st.empty() + + for chunk in response: + if chunk.choices[0].delta.reasoning_content: + reasoning_piece = chunk.choices[0].delta.reasoning_content + reasoning_content += reasoning_piece + reasoning_placeholder.markdown(reasoning_content) + elif chunk.choices[0].delta.content: + final_content += chunk.choices[0].delta.content - elapsed_time = time.time() - start_time - if elapsed_time >= 60: - time_str = f"{elapsed_time/60:.1f} minutes" - else: - time_str = f"{elapsed_time:.1f} seconds" - rprint(f"\n\n[yellow]Thought for {time_str}[/]") + elapsed_time = time.time() - start_time + time_str = f"{elapsed_time/60:.1f} minutes" if elapsed_time >= 60 else f"{elapsed_time:.1f} seconds" + st.caption(f"⏱️ Thought for {time_str}") - if self.show_reasoning: - print("\n") return reasoning_content def get_claude_response(self, user_input: str, reasoning: str) -> str: @@ -116,46 +90,46 @@ class ModelChain: } messages = [user_message, assistant_prefill] - response_placeholder = st.empty() - rprint(f"[green]{self.get_model_display_name()}[/]", end="") - try: - with self.claude_client.messages.stream( - model=self.current_model, - messages=messages, - max_tokens=8000 - ) as stream: - full_response = "" - for text in stream.text_stream: - full_response += text - response_placeholder.markdown(full_response) + # Create expander for Claude's response + with st.expander("🤖 Claude's Response", expanded=True): + response_placeholder = st.empty() + + with self.claude_client.messages.stream( + model=self.current_model, + messages=messages, + max_tokens=8000 + ) as stream: + full_response = "" + for text in stream.text_stream: + full_response += text + response_placeholder.markdown(full_response) - self.claude_messages.extend([user_message, { - "role": "assistant", - "content": [{"type": "text", "text": full_response}] - }]) - self.deepseek_messages.append({"role": "assistant", "content": full_response}) + self.claude_messages.extend([user_message, { + "role": "assistant", + "content": [{"type": "text", "text": full_response}] + }]) + self.deepseek_messages.append({"role": "assistant", "content": full_response}) - print("\n") - return full_response + return full_response except Exception as e: - rprint(f"\n[red]Error in response: {str(e)}[/]") + st.error(f"Error in response: {str(e)}") return "Error occurred while getting response" def main() -> None: """Main function to run the Streamlit app.""" - st.title("AI Assistant") + st.title("🤖 AI Assistant") # Sidebar for API keys with st.sidebar: - st.header("API Configuration") + st.header("⚙️ Configuration") deepseek_api_key = st.text_input("DeepSeek API Key", type="password") anthropic_api_key = st.text_input("Anthropic API Key", type="password") show_reasoning = st.toggle("Show Reasoning Process", value=True) - if st.button("Clear Chat History"): + if st.button("🗑️ Clear Chat History"): st.session_state.messages = [] st.experimental_rerun() @@ -171,7 +145,7 @@ def main() -> None: # Chat input if prompt := st.chat_input("What would you like to know?"): if not deepseek_api_key or not anthropic_api_key: - st.error("Please enter both API keys in the sidebar.") + st.error("⚠️ Please enter both API keys in the sidebar.") return # Initialize ModelChain @@ -185,12 +159,14 @@ def main() -> None: # Get AI response with st.chat_message("assistant"): if show_reasoning: - reasoning = chain.get_deepseek_reasoning(prompt) + with st.spinner("🤔 Thinking..."): + reasoning = chain.get_deepseek_reasoning(prompt) else: reasoning = "" - response = chain.get_claude_response(prompt, reasoning) - st.session_state.messages.append({"role": "assistant", "content": response}) + with st.spinner("✍️ Responding..."): + response = chain.get_claude_response(prompt, reasoning) + st.session_state.messages.append({"role": "assistant", "content": response}) if __name__ == "__main__": main() \ No newline at end of file From 2407a8ab6414448175d2873f0d777ab558dd0f45 Mon Sep 17 00:00:00 2001 From: Madhu Date: Tue, 28 Jan 2025 23:40:47 +0530 Subject: [PATCH 04/13] reasoning display --- .../ai_r1-tooluse-langroid/main.py | 62 +++----- .../ai_r1-tooluse-langroid/test.py | 147 ------------------ 2 files changed, 21 insertions(+), 188 deletions(-) delete mode 100644 ai_agent_tutorials/ai_r1-tooluse-langroid/test.py diff --git a/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py b/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py index e94023f..3a294f8 100644 --- a/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py +++ b/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py @@ -5,10 +5,6 @@ import streamlit as st from openai import OpenAI import anthropic from dotenv import load_dotenv -from rich import print as rprint -from rich.panel import Panel -from prompt_toolkit import PromptSession -from prompt_toolkit.styles import Style # Model Constants DEEPSEEK_MODEL: str = "deepseek-reasoner" @@ -36,49 +32,33 @@ class ModelChain: def get_model_display_name(self): return self.current_model - def get_deepseek_reasoning(self, user_input: str) -> str: + def get_deepseek_reasoning(self, user_input: str) -> str: start_time = time.time() self.deepseek_messages.append({"role": "user", "content": user_input}) - response = self.deepseek_client.chat.completions.create( - model=DEEPSEEK_MODEL, - max_tokens=1, - messages=self.deepseek_messages, - stream=True - ) - - reasoning_content = "" - final_content = "" - - # Create expander for reasoning - with st.expander("💭 Reasoning Process", expanded=True): - reasoning_placeholder = st.empty() + try: + response = self.deepseek_client.chat.completions.create( + max_tokens=1, # Keep max_tokens=1 to only get reasoning + model=DEEPSEEK_MODEL, + messages=self.deepseek_messages + ) - for chunk in response: - if chunk.choices[0].delta.reasoning_content: - reasoning_piece = chunk.choices[0].delta.reasoning_content - reasoning_content += reasoning_piece - reasoning_placeholder.markdown(reasoning_content) - elif chunk.choices[0].delta.content: - final_content += chunk.choices[0].delta.content + reasoning_content = response.choices[0].message.reasoning_content + + # Create expander for reasoning + with st.expander("💭 Reasoning Process", expanded=True): + st.markdown(reasoning_content) + elapsed_time = time.time() - start_time + time_str = f"{elapsed_time/60:.1f} minutes" if elapsed_time >= 60 else f"{elapsed_time:.1f} seconds" + st.caption(f"⏱️ Thought for {time_str}") - elapsed_time = time.time() - start_time - time_str = f"{elapsed_time/60:.1f} minutes" if elapsed_time >= 60 else f"{elapsed_time:.1f} seconds" - st.caption(f"⏱️ Thought for {time_str}") + return reasoning_content - return reasoning_content + except Exception as e: + st.error(f"Error getting DeepSeek reasoning: {str(e)}") + return "Error occurred while getting reasoning" def get_claude_response(self, user_input: str, reasoning: str) -> str: - """ - Get response from Claude model. - - Args: - user_input: User's input text - reasoning: Reasoning from DeepSeek - - Returns: - str: Claude's response - """ user_message = { "role": "user", "content": [{"type": "text", "text": user_input}] @@ -106,16 +86,16 @@ class ModelChain: full_response += text response_placeholder.markdown(full_response) + # Store the messages in Claude's history only self.claude_messages.extend([user_message, { "role": "assistant", "content": [{"type": "text", "text": full_response}] }]) - self.deepseek_messages.append({"role": "assistant", "content": full_response}) return full_response except Exception as e: - st.error(f"Error in response: {str(e)}") + st.error(f"Error in Claude response: {str(e)}") return "Error occurred while getting response" def main() -> None: diff --git a/ai_agent_tutorials/ai_r1-tooluse-langroid/test.py b/ai_agent_tutorials/ai_r1-tooluse-langroid/test.py deleted file mode 100644 index 0350d0d..0000000 --- a/ai_agent_tutorials/ai_r1-tooluse-langroid/test.py +++ /dev/null @@ -1,147 +0,0 @@ - -import os -from typing import List -import fire - -from langroid.pydantic_v1 import BaseModel, Field -import langroid as lr -from langroid.utils.configuration import settings -from langroid.agent.tool_message import ToolMessage -from langroid.agent.tools.orchestration import FinalResultTool -import langroid.language_models as lm -from rich.prompt import Prompt -from langroid.agent.chat_document import ChatDocument - -# for best results: -DEFAULT_LLM = lm.OpenAIChatModel.GPT4o - -os.environ["TOKENIZERS_PARALLELISM"] = "false" - -# (1) Define the desired structure via Pydantic. -# Here we define a nested structure for City information. -# The "Field" annotations are optional, and are included in the system message -# if provided, and help with generation accuracy. - - -class CityData(BaseModel): - population: int = Field(..., description="population of city") - country: str = Field(..., description="country of city") - - -class City(BaseModel): - name: str = Field(..., description="name of city") - details: CityData = Field(..., description="details of city") - - -# (2) Define the Tool class for the LLM to use, to produce the above structure. -class CityTool(lr.agent.ToolMessage): - """Present information about a city""" - - request: str = "city_tool" - purpose: str = """ - To present AFTER user gives a city name, - with all fields of the appropriate type filled out; - """ - city_info: City = Field(..., description="information about a city") - - def handle(self) -> FinalResultTool: - """Handle LLM's structured output if it matches City structure""" - print("SUCCESS! Got Valid City Info") - return FinalResultTool(answer=self.city_info) - - @staticmethod - def handle_message_fallback( - agent: lr.ChatAgent, msg: str | ChatDocument - ) -> str | None: - """ - We end up here when there was no recognized tool msg from the LLM; - In this case use the AgentDoneTool with content set to - the original message content. - """ - if isinstance(msg, ChatDocument) and msg.metadata.sender == lr.Entity.LLM: - return f""" - You forgot to use the TOOL/Function `{CityTool.name()}`. - Please use this tool to present the city info. - """ - - @classmethod - def examples(cls) -> List["ToolMessage"]: - # Used to provide few-shot examples in the system prompt - return [ - cls( - city_info=City( - name="San Francisco", - details=CityData( - population=800_000, - country="USA", - ), - ) - ) - ] - - -def app( - m: str = DEFAULT_LLM, # model - d: bool = False, # pass -d to enable debug mode (see prompts etc) - nc: bool = False, # pass -nc to disable cache-retrieval (i.e. get fresh answers) -): - settings.debug = d - settings.cache = not nc - # create LLM config - llm_cfg = lm.OpenAIGPTConfig( - chat_model=m or DEFAULT_LLM, - chat_context_length=32000, # set this based on model - max_output_tokens=1000, - temperature=0.2, - stream=True, - timeout=45, - ) - - # Recommended: First test if basic chat works with this llm setup as below: - # Once this works, then you can try the rest of the example. - # - # agent = lr.ChatAgent( - # lr.ChatAgentConfig( - # llm=llm_cfg, - # ) - # ) - # - # agent.llm_response("What is 3 + 4?") - # - # task = lr.Task(agent) - # verify you can interact with this in a chat loop on cmd line: - # task.run("Concisely answer some questions") - - # Define a ChatAgentConfig and ChatAgent - - config = lr.ChatAgentConfig( - llm=llm_cfg, - system_message=f""" - You will receive a city name, - and you must use the TOOL/FUNCTION `{CityTool.name()}` to generate/present - information about the city. - """, - ) - - agent = lr.ChatAgent(config) - - # (4) Enable the Tool for this agent --> this auto-inserts JSON instructions - # and few-shot examples (specified in the tool defn above) into the system message - agent.enable_message(CityTool) - - # (5) Create task specialized to return City object - task: City | None = lr.Task(agent, interactive=False)[City] - - while True: - city = Prompt.ask("Enter a city name") - if city in ["q", "x"]: - break - result: City | None = task.run(city) - if result: - print(f"City Info: {result}") - else: - print("No valid city info found.") - - -if __name__ == "__main__": - fire.Fire(app) \ No newline at end of file From a11f16f5fceeae0c62bd214ce0aec00ee94089f8 Mon Sep 17 00:00:00 2001 From: Madhu Date: Tue, 28 Jan 2025 23:56:41 +0530 Subject: [PATCH 05/13] fixes --- .../ai_r1-tooluse-langroid/main.py | 33 +++++++++++-------- 1 file changed, 20 insertions(+), 13 deletions(-) diff --git a/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py b/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py index 3a294f8..067d6e6 100644 --- a/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py +++ b/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py @@ -5,6 +5,7 @@ import streamlit as st from openai import OpenAI import anthropic from dotenv import load_dotenv +import json # Model Constants DEEPSEEK_MODEL: str = "deepseek-reasoner" @@ -15,9 +16,9 @@ load_dotenv() class ModelChain: def __init__(self, deepseek_api_key: str, anthropic_api_key: str) -> None: - self.deepseek_client = OpenAI( + self.client = OpenAI( api_key=deepseek_api_key, - base_url="https://api.deepseek.com" + base_url="https://api.deepseek.com" # Added /v1 to the base URL ) self.claude_client = anthropic.Anthropic(api_key=anthropic_api_key) @@ -26,24 +27,28 @@ class ModelChain: self.current_model: str = CLAUDE_MODEL self.show_reasoning = True - def set_model(self, model_name): - self.current_model = model_name - - def get_model_display_name(self): - return self.current_model - def get_deepseek_reasoning(self, user_input: str) -> str: start_time = time.time() - self.deepseek_messages.append({"role": "user", "content": user_input}) try: - response = self.deepseek_client.chat.completions.create( - max_tokens=1, # Keep max_tokens=1 to only get reasoning + # Debug print + st.write("Sending request to DeepSeek with messages:", json.dumps(self.deepseek_messages, indent=2)) + + deepseek_response = self.client.chat.completions.create( model=DEEPSEEK_MODEL, - messages=self.deepseek_messages + messages=[ + {"role": "system", "content": "You are an expert at reasoning and thinking from first principles."}, + {"role": "user", "content": user_input} + ], + max_tokens=1, + temperature=0.7, # Added temperature + stream=False # Explicitly set stream to False ) - reasoning_content = response.choices[0].message.reasoning_content + # Debug print + st.write("Raw response from DeepSeek:", deepseek_response) + + reasoning_content = deepseek_response.choices[0].message.reasoning_content # Create expander for reasoning with st.expander("💭 Reasoning Process", expanded=True): @@ -56,6 +61,8 @@ class ModelChain: except Exception as e: st.error(f"Error getting DeepSeek reasoning: {str(e)}") + st.error("Full error details:") + st.exception(e) return "Error occurred while getting reasoning" def get_claude_response(self, user_input: str, reasoning: str) -> str: From 00f4d4e8616e07a63dbf49d24cca91d4aa8c134d Mon Sep 17 00:00:00 2001 From: Madhu Date: Tue, 28 Jan 2025 23:58:42 +0530 Subject: [PATCH 06/13] fixes in deepseek api call --- ai_agent_tutorials/ai_r1-tooluse-langroid/main.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py b/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py index 067d6e6..6a70e1d 100644 --- a/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py +++ b/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py @@ -41,12 +41,11 @@ class ModelChain: {"role": "user", "content": user_input} ], max_tokens=1, - temperature=0.7, # Added temperature stream=False # Explicitly set stream to False ) # Debug print - st.write("Raw response from DeepSeek:", deepseek_response) + st.write("Raw response by DeepSeek:", deepseek_response) reasoning_content = deepseek_response.choices[0].message.reasoning_content From b6201eb82d6c15d7f6253c6acafa6e60fdfc9cf9 Mon Sep 17 00:00:00 2001 From: Madhu Date: Wed, 29 Jan 2025 00:50:41 +0530 Subject: [PATCH 07/13] api is down :/ --- .../ai_r1-tooluse-langroid/main.py | 29 +++++++------------ 1 file changed, 10 insertions(+), 19 deletions(-) diff --git a/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py b/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py index 6a70e1d..c70e57b 100644 --- a/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py +++ b/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py @@ -18,35 +18,29 @@ class ModelChain: def __init__(self, deepseek_api_key: str, anthropic_api_key: str) -> None: self.client = OpenAI( api_key=deepseek_api_key, - base_url="https://api.deepseek.com" # Added /v1 to the base URL + base_url="https://api.deepseek.com" ) self.claude_client = anthropic.Anthropic(api_key=anthropic_api_key) self.deepseek_messages: List[Dict[str, str]] = [] self.claude_messages: List[Dict[str, Any]] = [] self.current_model: str = CLAUDE_MODEL - self.show_reasoning = True def get_deepseek_reasoning(self, user_input: str) -> str: start_time = time.time() try: - # Debug print - st.write("Sending request to DeepSeek with messages:", json.dumps(self.deepseek_messages, indent=2)) - + deepseek_response = self.client.chat.completions.create( - model=DEEPSEEK_MODEL, + model="deepseek-reasoner", messages=[ {"role": "system", "content": "You are an expert at reasoning and thinking from first principles."}, {"role": "user", "content": user_input} ], max_tokens=1, - stream=False # Explicitly set stream to False + stream=False ) - - # Debug print - st.write("Raw response by DeepSeek:", deepseek_response) - + reasoning_content = deepseek_response.choices[0].message.reasoning_content # Create expander for reasoning @@ -106,18 +100,17 @@ class ModelChain: def main() -> None: """Main function to run the Streamlit app.""" - st.title("🤖 AI Assistant") + st.title("🤖 AI Project with deepseek + R1") # Sidebar for API keys with st.sidebar: st.header("⚙️ Configuration") deepseek_api_key = st.text_input("DeepSeek API Key", type="password") anthropic_api_key = st.text_input("Anthropic API Key", type="password") - show_reasoning = st.toggle("Show Reasoning Process", value=True) if st.button("🗑️ Clear Chat History"): st.session_state.messages = [] - st.experimental_rerun() + st.rerun() # Initialize session state for messages if "messages" not in st.session_state: @@ -144,11 +137,9 @@ def main() -> None: # Get AI response with st.chat_message("assistant"): - if show_reasoning: - with st.spinner("🤔 Thinking..."): - reasoning = chain.get_deepseek_reasoning(prompt) - else: - reasoning = "" + with st.spinner("🤔 Thinking..."): + reasoning = chain.get_deepseek_reasoning(prompt) + with st.spinner("✍️ Responding..."): response = chain.get_claude_response(prompt, reasoning) From 125b29ca1de6fa7439a9cb8edf3518df59c639cb Mon Sep 17 00:00:00 2001 From: Madhu Date: Wed, 29 Jan 2025 16:57:05 +0530 Subject: [PATCH 08/13] pydantic schema inclusions --- .../ai_r1-tooluse-langroid/README.md | 35 +++ .../ai_r1-tooluse-langroid/main.py | 208 ++++++++++++++++-- .../ai_r1-tooluse-langroid/test.py | 0 3 files changed, 225 insertions(+), 18 deletions(-) create mode 100644 ai_agent_tutorials/ai_r1-tooluse-langroid/test.py diff --git a/ai_agent_tutorials/ai_r1-tooluse-langroid/README.md b/ai_agent_tutorials/ai_r1-tooluse-langroid/README.md index e69de29..6d87421 100644 --- a/ai_agent_tutorials/ai_r1-tooluse-langroid/README.md +++ b/ai_agent_tutorials/ai_r1-tooluse-langroid/README.md @@ -0,0 +1,35 @@ +## Example Test Prompts + +### 1. Real-time Event Processing System +"We're building a real-time event processing system for a smart city infrastructure that needs to handle 100,000 IoT sensors, process environmental data, manage traffic flows, and provide emergency response capabilities. The system needs to be highly available, handle peak loads during emergencies, and maintain data integrity. Budget constraints exist but reliability is critical." + +### 2. Healthcare Data Platform +"Design a HIPAA-compliant healthcare data platform that needs to integrate with legacy systems, handle real-time patient monitoring, support ML-based diagnostics, and manage secure data sharing between different healthcare providers. The system should scale to handle data from 50 hospitals and support both real-time analytics and batch processing." + +### 3. Financial Trading Platform +"We need to build a high-frequency trading platform that processes market data streams, executes trades with sub-millisecond latency, maintains audit trails, and handles complex risk calculations. The system needs to be globally distributed, handle 100,000 transactions per second, and have robust disaster recovery capabilities." + +### 4. Multi-tenant SaaS Platform +"Design a multi-tenant SaaS platform for enterprise resource planning that needs to support customization per tenant, handle different data residency requirements, support offline capabilities, and maintain performance isolation between tenants. The system should scale to 10,000 concurrent users and support custom integrations." + +### 5. Digital Content Delivery Network +"We're building a global content delivery platform for streaming high-definition video content, supporting live streaming, VOD, and interactive content. The system needs to handle dynamic transcoding, support DRM, manage user-generated content, and optimize delivery based on network conditions and device capabilities." + +### 6. Supply Chain Management System +"Design a blockchain-based supply chain management system that needs to track products from source to retail, integrate with IoT sensors for condition monitoring, support smart contracts for automated settlements, and provide real-time visibility across the supply chain. The system should handle 1000 partners and support regulatory compliance reporting." + +Each of these prompts presents complex architectural challenges that require careful consideration of: +- Scalability patterns +- Data consistency requirements +- Security and compliance needs +- Integration complexities +- Performance optimization +- Cost-benefit trade-offs +- Technical debt implications +- Team expertise requirements + +The DeepSeek model will analyze these requirements and provide structured recommendations using the ProjectAnalysis schema, which Claude can then use to provide detailed implementation guidance. +Startups making critical technical decisions +Enterprise architecture modernization projects + + diff --git a/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py b/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py index c70e57b..f7086f5 100644 --- a/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py +++ b/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py @@ -1,10 +1,12 @@ -from typing import Optional, List, Dict, Any +from typing import Optional, List, Dict, Any, Union import os import time import streamlit as st from openai import OpenAI import anthropic from dotenv import load_dotenv +from pydantic import BaseModel, Field +from enum import Enum import json # Model Constants @@ -14,6 +16,162 @@ CLAUDE_MODEL: str = "claude-3-5-sonnet-20241022" # Load environment variables load_dotenv() +class ArchitecturePattern(str, Enum): + MICROSERVICES = "microservices" + MONOLITHIC = "monolithic" + SERVERLESS = "serverless" + EVENT_DRIVEN = "event_driven" + LAYERED = "layered" + +class SecurityLevel(str, Enum): + LOW = "low" + MEDIUM = "medium" + HIGH = "high" + VERY_HIGH = "very_high" + +class ScalabilityRequirement(str, Enum): + SMALL = "small" + MEDIUM = "medium" + LARGE = "large" + ENTERPRISE = "enterprise" + +class DatabaseType(str, Enum): + SQL = "sql" + NOSQL = "nosql" + GRAPH = "graph" + TIME_SERIES = "time_series" + HYBRID = "hybrid" + +class DevTool(BaseModel): + name: str + purpose: str + complexity: int = Field(ge=1, le=10) + setup_time_minutes: int + learning_curve: int = Field(ge=1, le=10) + alternatives: List[str] + +class InfrastructureComponent(BaseModel): + service_name: str + provider: str + estimated_cost: float + scaling_capability: ScalabilityRequirement + region: Optional[str] + backup_strategy: Optional[str] + +class ComplianceStandard(str, Enum): + HIPAA = "hipaa" + GDPR = "gdpr" + SOC2 = "soc2" + HITECH = "hitech" + ISO27001 = "iso27001" + PCI_DSS = "pci_dss" + +class DataClassification(str, Enum): + PHI = "protected_health_information" + PII = "personally_identifiable_information" + CONFIDENTIAL = "confidential" + PUBLIC = "public" + +class IntegrationType(str, Enum): + HL7 = "hl7" + FHIR = "fhir" + DICOM = "dicom" + REST = "rest" + SOAP = "soap" + CUSTOM = "custom" + +class DataProcessingType(str, Enum): + REAL_TIME = "real_time" + BATCH = "batch" + HYBRID = "hybrid" + +class MLCapability(BaseModel): + """Defines machine learning capabilities and requirements""" + model_type: str = Field(..., description="Type of ML model (e.g., diagnostic, predictive, monitoring)") + training_frequency: str = Field(..., description="How often the model needs retraining") + input_data_types: List[str] = Field(..., description="Types of data the model processes") + performance_requirements: Dict[str, float] = Field(..., description="Required metrics like accuracy, latency") + hardware_requirements: Dict[str, str] = Field(..., description="GPU/CPU/Memory requirements") + regulatory_constraints: List[str] = Field(..., description="Regulatory requirements for ML models") + +class DataIntegration(BaseModel): + """Defines integration points with external systems""" + system_name: str + integration_type: IntegrationType + data_frequency: str = Field(..., description="Frequency of data exchange") + data_volume: str = Field(..., description="Expected data volume per time unit") + transformation_rules: List[str] = Field(..., description="Data transformation requirements") + error_handling: Dict[str, str] = Field(..., description="Error handling strategies") + fallback_mechanism: Optional[str] = Field(None, description="Fallback approach when integration fails") + +class SecurityMeasure(BaseModel): + """Enhanced security measures for healthcare systems""" + measure_type: str + implementation_priority: int = Field(ge=1, le=5, description="Priority level for implementation") + compliance_standards: List[ComplianceStandard] + estimated_setup_time_days: int + data_classification: DataClassification + encryption_requirements: Dict[str, str] = Field(..., description="Encryption requirements for different states") + access_control_policy: Dict[str, List[str]] = Field(..., description="Role-based access control definitions") + audit_requirements: List[str] = Field(..., description="Audit logging requirements") + +class PerformanceRequirement(BaseModel): + """System performance requirements""" + metric_name: str = Field(..., description="Name of the performance metric") + threshold: float = Field(..., description="Required threshold value") + measurement_unit: str = Field(..., description="Unit of measurement") + criticality: int = Field(ge=1, le=5, description="How critical is this metric") + monitoring_frequency: str = Field(..., description="How often to monitor this metric") + +class ArchitectureDecision(BaseModel): + pattern: ArchitecturePattern + reasoning: str + trade_offs: Dict[str, List[str]] + estimated_implementation_time_months: float + +class TechnicalDebtItem(BaseModel): + description: str + severity: int = Field(ge=1, le=5) + estimated_fix_time_days: int + affected_components: List[str] + potential_risks: List[str] + +class ProjectAnalysis(BaseModel): + """Enhanced project analysis for healthcare systems""" + architecture_decision: ArchitectureDecision + recommended_tools: List[DevTool] + infrastructure: List[InfrastructureComponent] + security_measures: List[SecurityMeasure] + database_choice: DatabaseType + technical_debt_assessment: List[TechnicalDebtItem] + estimated_team_size: int + critical_path_components: List[str] + risk_assessment: Dict[str, str] + maintenance_considerations: List[str] + + # New healthcare-specific fields + compliance_requirements: List[ComplianceStandard] = Field( + ..., description="Required compliance standards" + ) + data_integrations: List[DataIntegration] = Field( + ..., description="External system integrations" + ) + ml_capabilities: List[MLCapability] = Field( + ..., description="ML model requirements and capabilities" + ) + performance_requirements: List[PerformanceRequirement] = Field( + ..., description="System performance requirements" + ) + data_retention_policy: Dict[str, str] = Field( + ..., description="Data retention requirements by type" + ) + disaster_recovery: Dict[str, Any] = Field( + ..., description="Disaster recovery and business continuity plans" + ) + interoperability_standards: List[str] = Field( + ..., description="Required healthcare interoperability standards" + ) + class ModelChain: def __init__(self, deepseek_api_key: str, anthropic_api_key: str) -> None: self.client = OpenAI( @@ -29,36 +187,51 @@ class ModelChain: def get_deepseek_reasoning(self, user_input: str) -> str: start_time = time.time() + system_prompt = """You are an expert software architect and technical advisor. Analyze the user's project requirements + and provide structured reasoning about architecture, tools, and implementation strategies. Your output must be a valid + JSON that matches the ProjectAnalysis schema. Consider scalability, security, maintenance, and technical debt in your analysis. + Focus on practical, modern solutions while being mindful of trade-offs.""" + try: - deepseek_response = self.client.chat.completions.create( model="deepseek-reasoner", messages=[ - {"role": "system", "content": "You are an expert at reasoning and thinking from first principles."}, + {"role": "system", "content": system_prompt}, {"role": "user", "content": user_input} ], - max_tokens=1, + max_tokens=3000, stream=False ) reasoning_content = deepseek_response.choices[0].message.reasoning_content - # Create expander for reasoning - with st.expander("💭 Reasoning Process", expanded=True): - st.markdown(reasoning_content) - elapsed_time = time.time() - start_time - time_str = f"{elapsed_time/60:.1f} minutes" if elapsed_time >= 60 else f"{elapsed_time:.1f} seconds" - st.caption(f"⏱️ Thought for {time_str}") + # Validate the reasoning content as ProjectAnalysis + try: + project_analysis = ProjectAnalysis.parse_raw(reasoning_content) + formatted_reasoning = json.dumps(json.loads(reasoning_content), indent=2) + + with st.expander("💭 Technical Analysis", expanded=True): + st.json(formatted_reasoning) + elapsed_time = time.time() - start_time + time_str = f"{elapsed_time/60:.1f} minutes" if elapsed_time >= 60 else f"{elapsed_time:.1f} seconds" + st.caption(f"⏱️ Analysis completed in {time_str}") - return reasoning_content + return reasoning_content + + except Exception as validation_error: + st.error(f"Invalid analysis format: {str(validation_error)}") + return "Error in analysis format" except Exception as e: - st.error(f"Error getting DeepSeek reasoning: {str(e)}") - st.error("Full error details:") - st.exception(e) - return "Error occurred while getting reasoning" + st.error(f"Error in DeepSeek analysis: {str(e)}") + return "Error occurred while analyzing" def get_claude_response(self, user_input: str, reasoning: str) -> str: + system_prompt = """You are a senior software architect and implementation advisor. Using the provided technical analysis, + give detailed, actionable advice for implementing the solution. Include code snippets, configuration examples, and + step-by-step implementation guidelines where appropriate. Focus on practical implementation details while maintaining + best practices and addressing potential challenges.""" + user_message = { "role": "user", "content": [{"type": "text", "text": user_input}] @@ -69,7 +242,7 @@ class ModelChain: "content": [{"type": "text", "text": f"{reasoning}"}] } - messages = [user_message, assistant_prefill] + messages = [assistant_prefill] try: # Create expander for Claude's response @@ -86,7 +259,6 @@ class ModelChain: full_response += text response_placeholder.markdown(full_response) - # Store the messages in Claude's history only self.claude_messages.extend([user_message, { "role": "assistant", "content": [{"type": "text", "text": full_response}] @@ -100,7 +272,7 @@ class ModelChain: def main() -> None: """Main function to run the Streamlit app.""" - st.title("🤖 AI Project with deepseek + R1") + st.title("🤖 AI Project with Deepseek + R1") # Sidebar for API keys with st.sidebar: diff --git a/ai_agent_tutorials/ai_r1-tooluse-langroid/test.py b/ai_agent_tutorials/ai_r1-tooluse-langroid/test.py new file mode 100644 index 0000000..e69de29 From b6d2a8ab05edc4b3823b0a06e51f810de729349e Mon Sep 17 00:00:00 2001 From: Madhu Date: Wed, 29 Jan 2025 21:18:02 +0530 Subject: [PATCH 09/13] complete schemas --- .../ai_r1-tooluse-langroid/main.py | 232 +++++++++++------- .../ai_r1-tooluse-langroid/test.py | 0 2 files changed, 148 insertions(+), 84 deletions(-) delete mode 100644 ai_agent_tutorials/ai_r1-tooluse-langroid/test.py diff --git a/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py b/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py index f7086f5..8034005 100644 --- a/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py +++ b/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py @@ -18,7 +18,7 @@ load_dotenv() class ArchitecturePattern(str, Enum): MICROSERVICES = "microservices" - MONOLITHIC = "monolithic" + MONOLITHIC = "monolithic" SERVERLESS = "serverless" EVENT_DRIVEN = "event_driven" LAYERED = "layered" @@ -42,22 +42,6 @@ class DatabaseType(str, Enum): TIME_SERIES = "time_series" HYBRID = "hybrid" -class DevTool(BaseModel): - name: str - purpose: str - complexity: int = Field(ge=1, le=10) - setup_time_minutes: int - learning_curve: int = Field(ge=1, le=10) - alternatives: List[str] - -class InfrastructureComponent(BaseModel): - service_name: str - provider: str - estimated_cost: float - scaling_capability: ScalabilityRequirement - region: Optional[str] - backup_strategy: Optional[str] - class ComplianceStandard(str, Enum): HIPAA = "hipaa" GDPR = "gdpr" @@ -94,16 +78,6 @@ class MLCapability(BaseModel): hardware_requirements: Dict[str, str] = Field(..., description="GPU/CPU/Memory requirements") regulatory_constraints: List[str] = Field(..., description="Regulatory requirements for ML models") -class DataIntegration(BaseModel): - """Defines integration points with external systems""" - system_name: str - integration_type: IntegrationType - data_frequency: str = Field(..., description="Frequency of data exchange") - data_volume: str = Field(..., description="Expected data volume per time unit") - transformation_rules: List[str] = Field(..., description="Data transformation requirements") - error_handling: Dict[str, str] = Field(..., description="Error handling strategies") - fallback_mechanism: Optional[str] = Field(None, description="Fallback approach when integration fails") - class SecurityMeasure(BaseModel): """Enhanced security measures for healthcare systems""" measure_type: str @@ -115,62 +89,78 @@ class SecurityMeasure(BaseModel): access_control_policy: Dict[str, List[str]] = Field(..., description="Role-based access control definitions") audit_requirements: List[str] = Field(..., description="Audit logging requirements") -class PerformanceRequirement(BaseModel): - """System performance requirements""" - metric_name: str = Field(..., description="Name of the performance metric") - threshold: float = Field(..., description="Required threshold value") - measurement_unit: str = Field(..., description="Unit of measurement") - criticality: int = Field(ge=1, le=5, description="How critical is this metric") - monitoring_frequency: str = Field(..., description="How often to monitor this metric") - class ArchitectureDecision(BaseModel): + """Architecture decision details""" pattern: ArchitecturePattern - reasoning: str + rationale: str trade_offs: Dict[str, List[str]] - estimated_implementation_time_months: float + estimated_cost: Dict[str, float] -class TechnicalDebtItem(BaseModel): - description: str - severity: int = Field(ge=1, le=5) - estimated_fix_time_days: int - affected_components: List[str] - potential_risks: List[str] +class InfrastructureResource(BaseModel): + """Infrastructure resource requirements""" + resource_type: str + specifications: Dict[str, str] + scaling_policy: Dict[str, Any] + estimated_cost: float + +class DataIntegration(BaseModel): + """Data integration specifications""" + integration_type: IntegrationType + data_format: str + frequency: str + volume: str + security_requirements: Dict[str, str] + +class PerformanceRequirement(BaseModel): + """Performance requirements specification""" + metric_name: str + target_value: float + measurement_unit: str + priority: int + +class AuditConfig(BaseModel): + """Audit configuration settings""" + log_retention_period: int + audit_events: List[str] + compliance_mapping: Dict[str, List[str]] + +class APIConfig(BaseModel): + """API configuration settings""" + version: str + auth_method: str + rate_limits: Dict[str, int] + documentation_url: str + +class ErrorHandlingConfig(BaseModel): + """Error handling configuration""" + retry_policy: Dict[str, Any] + fallback_strategies: List[str] + notification_channels: List[str] class ProjectAnalysis(BaseModel): """Enhanced project analysis for healthcare systems""" architecture_decision: ArchitectureDecision - recommended_tools: List[DevTool] - infrastructure: List[InfrastructureComponent] + infrastructure_resources: List[InfrastructureResource] security_measures: List[SecurityMeasure] database_choice: DatabaseType - technical_debt_assessment: List[TechnicalDebtItem] estimated_team_size: int critical_path_components: List[str] risk_assessment: Dict[str, str] maintenance_considerations: List[str] - # New healthcare-specific fields - compliance_requirements: List[ComplianceStandard] = Field( - ..., description="Required compliance standards" - ) - data_integrations: List[DataIntegration] = Field( - ..., description="External system integrations" - ) - ml_capabilities: List[MLCapability] = Field( - ..., description="ML model requirements and capabilities" - ) - performance_requirements: List[PerformanceRequirement] = Field( - ..., description="System performance requirements" - ) - data_retention_policy: Dict[str, str] = Field( - ..., description="Data retention requirements by type" - ) - disaster_recovery: Dict[str, Any] = Field( - ..., description="Disaster recovery and business continuity plans" - ) - interoperability_standards: List[str] = Field( - ..., description="Required healthcare interoperability standards" - ) + # Healthcare-specific fields + compliance_requirements: List[ComplianceStandard] + data_integrations: List[DataIntegration] + ml_capabilities: List[MLCapability] + performance_requirements: List[PerformanceRequirement] + data_retention_policy: Dict[str, str] + disaster_recovery: Dict[str, Any] + interoperability_standards: List[str] + + # New fields + audit_config: AuditConfig + api_config: APIConfig + error_handling: ErrorHandlingConfig class ModelChain: def __init__(self, deepseek_api_key: str, anthropic_api_key: str) -> None: @@ -183,13 +173,88 @@ class ModelChain: self.deepseek_messages: List[Dict[str, str]] = [] self.claude_messages: List[Dict[str, Any]] = [] self.current_model: str = CLAUDE_MODEL - def get_deepseek_reasoning(self, user_input: str) -> str: start_time = time.time() system_prompt = """You are an expert software architect and technical advisor. Analyze the user's project requirements - and provide structured reasoning about architecture, tools, and implementation strategies. Your output must be a valid - JSON that matches the ProjectAnalysis schema. Consider scalability, security, maintenance, and technical debt in your analysis. + and provide structured reasoning about architecture, tools, and implementation strategies. + + IMPORTANT: Your response must be a valid JSON object (not a string or any other format) that matches the schema provided below. + Do not include any explanatory text, markdown formatting, or code blocks - only return the JSON object. + + Schema: + { + "architecture_decision": { + "pattern": "one of: microservices|monolithic|serverless|event_driven|layered", + "rationale": "string", + "trade_offs": {"advantage": ["list of strings"], "disadvantage": ["list of strings"]}, + "estimated_cost": {"implementation": float, "maintenance": float} + }, + "infrastructure_resources": [{ + "resource_type": "string", + "specifications": {"key": "value"}, + "scaling_policy": {"key": "value"}, + "estimated_cost": float + }], + "security_measures": [{ + "measure_type": "string", + "implementation_priority": "integer 1-5", + "compliance_standards": ["hipaa", "gdpr", "soc2", "hitech", "iso27001", "pci_dss"], + "estimated_setup_time_days": "integer", + "data_classification": "one of: protected_health_information|personally_identifiable_information|confidential|public", + "encryption_requirements": {"key": "value"}, + "access_control_policy": {"role": ["permissions"]}, + "audit_requirements": ["list of strings"] + }], + "database_choice": "one of: sql|nosql|graph|time_series|hybrid", + "ml_capabilities": [{ + "model_type": "string", + "training_frequency": "string", + "input_data_types": ["list of strings"], + "performance_requirements": {"metric": float}, + "hardware_requirements": {"resource": "specification"}, + "regulatory_constraints": ["list of strings"] + }], + "data_integrations": [{ + "integration_type": "one of: hl7|fhir|dicom|rest|soap|custom", + "data_format": "string", + "frequency": "string", + "volume": "string", + "security_requirements": {"key": "value"} + }], + "performance_requirements": [{ + "metric_name": "string", + "target_value": float, + "measurement_unit": "string", + "priority": "integer 1-5" + }], + "audit_config": { + "log_retention_period": "integer", + "audit_events": ["list of strings"], + "compliance_mapping": {"standard": ["requirements"]} + }, + "api_config": { + "version": "string", + "auth_method": "string", + "rate_limits": {"role": "requests_per_minute"}, + "documentation_url": "string" + }, + "error_handling": { + "retry_policy": {"key": "value"}, + "fallback_strategies": ["list of strings"], + "notification_channels": ["list of strings"] + }, + "estimated_team_size": "integer", + "critical_path_components": ["list of strings"], + "risk_assessment": {"risk": "mitigation"}, + "maintenance_considerations": ["list of strings"], + "compliance_requirements": ["list of compliance standards"], + "data_retention_policy": {"data_type": "retention_period"}, + "disaster_recovery": {"key": "value"}, + "interoperability_standards": ["list of strings"] + } + + Consider scalability, security, maintenance, and technical debt in your analysis. Focus on practical, modern solutions while being mindful of trade-offs.""" try: @@ -204,24 +269,22 @@ class ModelChain: ) reasoning_content = deepseek_response.choices[0].message.reasoning_content + normal_content = deepseek_response.choices[0].message.content + + # Display the reasoning separately + with st.expander("DeepSeek Reasoning", expanded=True): + st.markdown(reasoning_content) - # Validate the reasoning content as ProjectAnalysis - try: - project_analysis = ProjectAnalysis.parse_raw(reasoning_content) - formatted_reasoning = json.dumps(json.loads(reasoning_content), indent=2) - with st.expander("💭 Technical Analysis", expanded=True): - st.json(formatted_reasoning) - elapsed_time = time.time() - start_time - time_str = f"{elapsed_time/60:.1f} minutes" if elapsed_time >= 60 else f"{elapsed_time:.1f} seconds" - st.caption(f"⏱️ Analysis completed in {time_str}") + with st.expander("💭 Technical Analysis", expanded=True): + st.markdown(normal_content) + elapsed_time = time.time() - start_time + time_str = f"{elapsed_time/60:.1f} minutes" if elapsed_time >= 60 else f"{elapsed_time:.1f} seconds" + st.caption(f"⏱️ Analysis completed in {time_str}") + # Return the validated structured output for Claude return reasoning_content - except Exception as validation_error: - st.error(f"Invalid analysis format: {str(validation_error)}") - return "Error in analysis format" - except Exception as e: st.error(f"Error in DeepSeek analysis: {str(e)}") return "Error occurred while analyzing" @@ -251,6 +314,7 @@ class ModelChain: with self.claude_client.messages.stream( model=self.current_model, + system=system_prompt, messages=messages, max_tokens=8000 ) as stream: diff --git a/ai_agent_tutorials/ai_r1-tooluse-langroid/test.py b/ai_agent_tutorials/ai_r1-tooluse-langroid/test.py deleted file mode 100644 index e69de29..0000000 From acd4e36963e770cc85ad3b8db60f62f56271ba4c Mon Sep 17 00:00:00 2001 From: Madhu Date: Thu, 30 Jan 2025 01:05:58 +0530 Subject: [PATCH 10/13] addition of pydantic --- .../ai_r1-tooluse-langroid/main.py | 254 ++++++------------ .../ai_r1-tooluse-langroid/test.py | 184 +++++++++++++ 2 files changed, 266 insertions(+), 172 deletions(-) create mode 100644 ai_agent_tutorials/ai_r1-tooluse-langroid/test.py diff --git a/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py b/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py index 8034005..9be0460 100644 --- a/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py +++ b/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py @@ -8,6 +8,8 @@ from dotenv import load_dotenv from pydantic import BaseModel, Field from enum import Enum import json +from phi.agent import Agent, RunResponse +from phi.model.anthropic import Claude # Model Constants DEEPSEEK_MODEL: str = "deepseek-reasoner" @@ -16,151 +18,71 @@ CLAUDE_MODEL: str = "claude-3-5-sonnet-20241022" # Load environment variables load_dotenv() +system_prompt = """You are a Senior Software Expert and Technical Documentation Assistant. Your role is to analyze the structured JSON response from DeepSeek, which contains architectural and technical recommendations across various domains, along with the original user query describing the software system they want to build. + + The input consists of: + - The user's original query describing their software requirements + - A structured JSON response containing recommendations for architecture, security, infrastructure, compliance and other technical domains + + For each key-value pair in the JSON: + 1. Present the key and its corresponding value in a readable report format + 2. Format the information in a clear, organized way + 3. Do not add your own opinions or suggestions + 4. Do not modify or reinterpret the provided information + + Keep your responses factual and directly based on the JSON content provided.""" + class ArchitecturePattern(str, Enum): - MICROSERVICES = "microservices" - MONOLITHIC = "monolithic" - SERVERLESS = "serverless" - EVENT_DRIVEN = "event_driven" - LAYERED = "layered" - -class SecurityLevel(str, Enum): - LOW = "low" - MEDIUM = "medium" - HIGH = "high" - VERY_HIGH = "very_high" - -class ScalabilityRequirement(str, Enum): - SMALL = "small" - MEDIUM = "medium" - LARGE = "large" - ENTERPRISE = "enterprise" + """Architectural patterns for system design.""" + MICROSERVICES = "microservices" # Decomposed into small, independent services + MONOLITHIC = "monolithic" # Single, unified codebase + SERVERLESS = "serverless" # Function-as-a-Service architecture + EVENT_DRIVEN = "event_driven" # Asynchronous event-based communication class DatabaseType(str, Enum): - SQL = "sql" - NOSQL = "nosql" - GRAPH = "graph" - TIME_SERIES = "time_series" - HYBRID = "hybrid" + """Types of database systems.""" + SQL = "sql" # Relational databases with ACID properties + NOSQL = "nosql" # Non-relational databases for flexible schemas + HYBRID = "hybrid" # Combined SQL and NoSQL approach class ComplianceStandard(str, Enum): - HIPAA = "hipaa" - GDPR = "gdpr" - SOC2 = "soc2" - HITECH = "hitech" - ISO27001 = "iso27001" - PCI_DSS = "pci_dss" - -class DataClassification(str, Enum): - PHI = "protected_health_information" - PII = "personally_identifiable_information" - CONFIDENTIAL = "confidential" - PUBLIC = "public" - -class IntegrationType(str, Enum): - HL7 = "hl7" - FHIR = "fhir" - DICOM = "dicom" - REST = "rest" - SOAP = "soap" - CUSTOM = "custom" - -class DataProcessingType(str, Enum): - REAL_TIME = "real_time" - BATCH = "batch" - HYBRID = "hybrid" - -class MLCapability(BaseModel): - """Defines machine learning capabilities and requirements""" - model_type: str = Field(..., description="Type of ML model (e.g., diagnostic, predictive, monitoring)") - training_frequency: str = Field(..., description="How often the model needs retraining") - input_data_types: List[str] = Field(..., description="Types of data the model processes") - performance_requirements: Dict[str, float] = Field(..., description="Required metrics like accuracy, latency") - hardware_requirements: Dict[str, str] = Field(..., description="GPU/CPU/Memory requirements") - regulatory_constraints: List[str] = Field(..., description="Regulatory requirements for ML models") - -class SecurityMeasure(BaseModel): - """Enhanced security measures for healthcare systems""" - measure_type: str - implementation_priority: int = Field(ge=1, le=5, description="Priority level for implementation") - compliance_standards: List[ComplianceStandard] - estimated_setup_time_days: int - data_classification: DataClassification - encryption_requirements: Dict[str, str] = Field(..., description="Encryption requirements for different states") - access_control_policy: Dict[str, List[str]] = Field(..., description="Role-based access control definitions") - audit_requirements: List[str] = Field(..., description="Audit logging requirements") + """Regulatory compliance standards.""" + HIPAA = "hipaa" # Healthcare data protection + GDPR = "gdpr" # EU data privacy regulation + SOC2 = "soc2" # Service organization security controls + ISO27001 = "iso27001" # Information security management class ArchitectureDecision(BaseModel): - """Architecture decision details""" + """Represents architectural decisions and their justifications.""" pattern: ArchitecturePattern - rationale: str - trade_offs: Dict[str, List[str]] - estimated_cost: Dict[str, float] + rationale: str = Field(..., min_length=50) # Detailed explanation for the choice + trade_offs: Dict[str, List[str]] = Field(..., alias="trade_offs") # Pros and cons + estimated_cost: Dict[str, float] # Cost breakdown + +class SecurityMeasure(BaseModel): + """Security controls and implementation details.""" + measure_type: str # Type of security measure + implementation_priority: int = Field(..., ge=1, le=5) # Priority level 1-5 + compliance_standards: List[ComplianceStandard] # Applicable standards + data_classification: str # Data sensitivity level class InfrastructureResource(BaseModel): - """Infrastructure resource requirements""" - resource_type: str - specifications: Dict[str, str] - scaling_policy: Dict[str, Any] - estimated_cost: float + """Infrastructure components and specifications.""" + resource_type: str # Type of infrastructure resource + specifications: Dict[str, str] # Technical specifications + scaling_policy: Dict[str, str] # Scaling rules and thresholds + estimated_cost: float # Estimated cost per resource -class DataIntegration(BaseModel): - """Data integration specifications""" - integration_type: IntegrationType - data_format: str - frequency: str - volume: str - security_requirements: Dict[str, str] +class TechnicalAnalysis(BaseModel): + """Complete technical analysis of the system architecture.""" + architecture_decision: ArchitectureDecision # Core architecture choices + infrastructure_resources: List[InfrastructureResource] # Required resources + security_measures: List[SecurityMeasure] # Security controls + database_choice: DatabaseType # Database architecture + compliance_requirements: List[ComplianceStandard] = [] # Required standards + performance_requirements: List[Dict[str, Union[str, float]]] = [] # Performance metrics + risk_assessment: Dict[str, str] = {} # Identified risks and mitigations -class PerformanceRequirement(BaseModel): - """Performance requirements specification""" - metric_name: str - target_value: float - measurement_unit: str - priority: int - -class AuditConfig(BaseModel): - """Audit configuration settings""" - log_retention_period: int - audit_events: List[str] - compliance_mapping: Dict[str, List[str]] - -class APIConfig(BaseModel): - """API configuration settings""" - version: str - auth_method: str - rate_limits: Dict[str, int] - documentation_url: str - -class ErrorHandlingConfig(BaseModel): - """Error handling configuration""" - retry_policy: Dict[str, Any] - fallback_strategies: List[str] - notification_channels: List[str] - -class ProjectAnalysis(BaseModel): - """Enhanced project analysis for healthcare systems""" - architecture_decision: ArchitectureDecision - infrastructure_resources: List[InfrastructureResource] - security_measures: List[SecurityMeasure] - database_choice: DatabaseType - estimated_team_size: int - critical_path_components: List[str] - risk_assessment: Dict[str, str] - maintenance_considerations: List[str] - - # Healthcare-specific fields - compliance_requirements: List[ComplianceStandard] - data_integrations: List[DataIntegration] - ml_capabilities: List[MLCapability] - performance_requirements: List[PerformanceRequirement] - data_retention_policy: Dict[str, str] - disaster_recovery: Dict[str, Any] - interoperability_standards: List[str] - - # New fields - audit_config: AuditConfig - api_config: APIConfig - error_handling: ErrorHandlingConfig class ModelChain: def __init__(self, deepseek_api_key: str, anthropic_api_key: str) -> None: @@ -169,15 +91,22 @@ class ModelChain: base_url="https://api.deepseek.com" ) self.claude_client = anthropic.Anthropic(api_key=anthropic_api_key) + self.agent = Agent( + model=Claude(id="claude-3-5-sonnet-20241022", api_key=anthropic_api_key), + system_prompt=system_prompt, + markdown=True + ) self.deepseek_messages: List[Dict[str, str]] = [] self.claude_messages: List[Dict[str, Any]] = [] self.current_model: str = CLAUDE_MODEL - def get_deepseek_reasoning(self, user_input: str) -> str: + def get_deepseek_reasoning(self, user_input: str) -> tuple[str, str]: start_time = time.time() system_prompt = """You are an expert software architect and technical advisor. Analyze the user's project requirements and provide structured reasoning about architecture, tools, and implementation strategies. + + IMPORTANT: Reason why you are choosing a particular architecture pattern, database type, etc. for user understanding in your reasoning. IMPORTANT: Your response must be a valid JSON object (not a string or any other format) that matches the schema provided below. Do not include any explanatory text, markdown formatting, or code blocks - only return the JSON object. @@ -282,53 +211,34 @@ class ModelChain: time_str = f"{elapsed_time/60:.1f} minutes" if elapsed_time >= 60 else f"{elapsed_time:.1f} seconds" st.caption(f"⏱️ Analysis completed in {time_str}") - # Return the validated structured output for Claude - return reasoning_content + # Return both reasoning and normal content + return reasoning_content, normal_content except Exception as e: st.error(f"Error in DeepSeek analysis: {str(e)}") - return "Error occurred while analyzing" - - def get_claude_response(self, user_input: str, reasoning: str) -> str: - system_prompt = """You are a senior software architect and implementation advisor. Using the provided technical analysis, - give detailed, actionable advice for implementing the solution. Include code snippets, configuration examples, and - step-by-step implementation guidelines where appropriate. Focus on practical implementation details while maintaining - best practices and addressing potential challenges.""" - - user_message = { - "role": "user", - "content": [{"type": "text", "text": user_input}] - } - - assistant_prefill = { - "role": "assistant", - "content": [{"type": "text", "text": f"{reasoning}"}] - } - - messages = [assistant_prefill] + return "Error occurred while analyzing", "" + def get_claude_response(self, user_input: str, deepseek_output: tuple[str, str]) -> str: try: + reasoning_content, normal_content = deepseek_output + # Create expander for Claude's response with st.expander("🤖 Claude's Response", expanded=True): response_placeholder = st.empty() - with self.claude_client.messages.stream( - model=self.current_model, - system=system_prompt, - messages=messages, - max_tokens=8000 - ) as stream: - full_response = "" - for text in stream.text_stream: - full_response += text - response_placeholder.markdown(full_response) + # Prepare the message with user input, reasoning and normal output + message = f"""User Query: {user_input} - self.claude_messages.extend([user_message, { - "role": "assistant", - "content": [{"type": "text", "text": full_response}] - }]) + DeepSeek Reasoning: {reasoning_content} - return full_response + DeepSeek Technical Analysis: {normal_content}""" + + # Use Phi Agent to get response + response: RunResponse = self.agent.run( + message=message + ) + + return response.content except Exception as e: st.error(f"Error in Claude response: {str(e)}") @@ -374,11 +284,11 @@ def main() -> None: # Get AI response with st.chat_message("assistant"): with st.spinner("🤔 Thinking..."): - reasoning = chain.get_deepseek_reasoning(prompt) + deepseek_output = chain.get_deepseek_reasoning(prompt) with st.spinner("✍️ Responding..."): - response = chain.get_claude_response(prompt, reasoning) + response = chain.get_claude_response(prompt, deepseek_output) st.session_state.messages.append({"role": "assistant", "content": response}) if __name__ == "__main__": diff --git a/ai_agent_tutorials/ai_r1-tooluse-langroid/test.py b/ai_agent_tutorials/ai_r1-tooluse-langroid/test.py new file mode 100644 index 0000000..35a9c40 --- /dev/null +++ b/ai_agent_tutorials/ai_r1-tooluse-langroid/test.py @@ -0,0 +1,184 @@ +from enum import Enum +from typing import List, Dict, Union +from pydantic import BaseModel, Field, ValidationError +import streamlit as st +from openai import OpenAI +import anthropic +import json +import re +import os +from dotenv import load_dotenv +from phi.agent import Agent, RunResponse +from phi.model.anthropic import Claude + +load_dotenv() + +# -------------------------- +# Enums & Data Models +# -------------------------- +class ArchitecturePattern(str, Enum): + MICROSERVICES = "microservices" + MONOLITHIC = "monolithic" + SERVERLESS = "serverless" + EVENT_DRIVEN = "event_driven" + +class DatabaseType(str, Enum): + SQL = "sql" + NOSQL = "nosql" + HYBRID = "hybrid" + +class ComplianceStandard(str, Enum): + HIPAA = "hipaa" + GDPR = "gdpr" + SOC2 = "soc2" + ISO27001 = "iso27001" + +class ArchitectureDecision(BaseModel): + pattern: ArchitecturePattern + rationale: str = Field(..., min_length=50) + trade_offs: Dict[str, List[str]] = Field(..., alias="trade_offs") + estimated_cost: Dict[str, float] + +class SecurityMeasure(BaseModel): + measure_type: str + implementation_priority: int = Field(..., ge=1, le=5) + compliance_standards: List[ComplianceStandard] + data_classification: str + +class InfrastructureResource(BaseModel): + resource_type: str + specifications: Dict[str, str] + scaling_policy: Dict[str, str] + estimated_cost: float + +class TechnicalAnalysis(BaseModel): + architecture_decision: ArchitectureDecision + infrastructure_resources: List[InfrastructureResource] + security_measures: List[SecurityMeasure] + database_choice: DatabaseType + compliance_requirements: List[ComplianceStandard] = [] + performance_requirements: List[Dict[str, Union[str, float]]] = [] + risk_assessment: Dict[str, str] = {} + +# -------------------------- +# Core Implementation +# -------------------------- +class ArchitectureAnalyzer: + def __init__(self, deepseek_api_key: str, anthropic_api_key: str): + self.deepseek_client = OpenAI( + api_key=deepseek_api_key, + base_url="https://api.deepseek.com" + ) + self.claude_agent = Agent( + model=Claude( + id="claude-3-5-sonnet-20241022", + api_key=anthropic_api_key + ), + markdown=True, + ) + self.reasoning_content = "" + + self.deepseek_prompt = f"""Analyze software requirements and return JSON with: +{{ + "architecture_decision": {{ + "pattern": "{'|'.join([e.value for e in ArchitecturePattern])}", + "rationale": "technical justification", + "trade_offs": {{"pros": [], "cons": []}}, + "estimated_cost": {{"development": float, "maintenance": float}} + }}, + "infrastructure_resources": [{{"resource_type": "...", "specifications": {{}}, ...}}], + "security_measures": [{{"measure_type": "...", "priority": 1-5, ...}}], + "database_choice": "{'|'.join([e.value for e in DatabaseType])}", + "compliance_requirements": ["..."], + "performance_requirements": [{{"metric": "...", "target": float}}] +}}""" + + def _extract_json(self, text: str) -> dict: + try: + json_str = re.search(r'\{.*\}', text, re.DOTALL).group() + return json.loads(json_str) + except (AttributeError, json.JSONDecodeError) as e: + st.error(f"JSON extraction failed: {str(e)}") + st.text("Raw response:\n" + text) + raise + + def analyze_requirements(self, user_input: str) -> TechnicalAnalysis: + try: + response1 = self.deepseek_client.chat.completions.create( + model="deepseek-reasoner", + messages=[ + {"role": "system", "content": self.deepseek_prompt}, + {"role": "user", "content": user_input} + ], + temperature=0.2, + max_tokens=2000 + ) + self.reasoning_content = response1.choices[0].message.reasoning_content + json_data = self._extract_json(response1.choices[0].message.content) + return TechnicalAnalysis(**json_data) + + except ValidationError as e: + st.error(f"Validation error: {e.errors()}") + st.json(json_data) + raise + + def generate_report(self, analysis: TechnicalAnalysis) -> str: + report_prompt = f"""Convert this technical analysis into a executive report: +{analysis.model_dump_json(indent=2)} + +Use markdown with: +# Title +## Sections +- Bullet points +**Bold important items** +Tables for cost/performance""" + + response = self.claude_agent.run(report_prompt) + return response.content + +# -------------------------- +# Streamlit UI +# -------------------------- +def main(): + st.title("🏗️ AI Architecture Advisor") + + with st.sidebar: + st.header("🔑 Setup") + deepseek_api_key = st.text_input("DeepSeek Key", type="password") + anthropic_api_key = st.text_input("Claude Key", type="password") + + if "analysis" not in st.session_state: + st.session_state.analysis = None + + if prompt := st.chat_input("Describe your system requirements:"): + if not all([deepseek_api_key, anthropic_api_key]): + st.error("Missing API keys") + return + + analyzer = ArchitectureAnalyzer(deepseek_api_key, anthropic_api_key) + + with st.status("🔨 Processing...", expanded=True): + try: + # Analysis Phase + st.write("🧠 Analyzing requirements...") + analysis = analyzer.analyze_requirements(prompt) + st.session_state.analysis = analysis + with st.expander("reasoning"): + st.markdown(analyzer.reasoning_content) + + # Reporting Phase + st.write("📊 Generating report...") + report = analyzer.generate_report(analysis) + + # Display Results + st.success("Analysis complete!") + st.markdown(report) + + with st.expander("📁 Raw Analysis Data"): + st.json(analysis.model_dump_json()) + + except Exception as e: + st.error(f"Processing failed: {str(e)}") + +if __name__ == "__main__": + main() \ No newline at end of file From 6497f1b51580f9bddf93f8ddc18d0dbdca4f6659 Mon Sep 17 00:00:00 2001 From: Madhu Date: Thu, 30 Jan 2025 02:20:25 +0530 Subject: [PATCH 11/13] new file names + readme --- .../ai_system_architect_r1/README.md | 74 +++++ .../ai_system_architect_r1.py | 311 ++++++++++++++++++ .../ai_system_architect_r1/requirements.txt | 4 + 3 files changed, 389 insertions(+) create mode 100644 ai_agent_tutorials/ai_system_architect_r1/README.md create mode 100644 ai_agent_tutorials/ai_system_architect_r1/ai_system_architect_r1.py create mode 100644 ai_agent_tutorials/ai_system_architect_r1/requirements.txt diff --git a/ai_agent_tutorials/ai_system_architect_r1/README.md b/ai_agent_tutorials/ai_system_architect_r1/README.md new file mode 100644 index 0000000..4c3a314 --- /dev/null +++ b/ai_agent_tutorials/ai_system_architect_r1/README.md @@ -0,0 +1,74 @@ +# 🤖 AI System Architect Advisor with R1 + +A Streamlit application that provides expert software architecture analysis and recommendations using a dual-model approach combining DeepSeek R1's Reasoning and Claude. The system provides detailed technical analysis, implementation roadmaps, and architectural decisions for complex software systems. + +## Features + +- **Dual AI Model Architecture** + - **DeepSeek Reasoner**: Provides initial technical analysis and structured reasoning about architecture patterns, tools, and implementation strategies + - **Claude-3.5**: Generates detailed explanations, implementation roadmaps, and technical specifications based on DeepSeek's analysis + +- **Comprehensive Analysis Components** + - Architecture Pattern Selection + - Infrastructure Resource Planning + - Security Measures and Compliance + - Database Architecture + - Performance Requirements + - Cost Estimation + - Risk Assessment + +- **Analysis Types** + - Real-time Event Processing Systems + - Healthcare Data Platforms + - Financial Trading Platforms + - Multi-tenant SaaS Solutions + - Digital Content Delivery Networks + - Supply Chain Management Systems + +## How to Run + +1. **Setup Environment** + ```bash + # Clone the repository + git clone https://github.com/Shubhamsaboo/awesome-llm-apps.git + cd awesome-llm-apps/ai_agent_tutorials/ai_system_architect_r1 + + # Install dependencies + pip install -r requirements.txt + ``` + +2. **Configure API Keys** + - Get DeepSeek API key from DeepSeek platform + - Get Anthropic API key from [Anthropic Platform](https://www.anthropic.com) + +3. **Run the Application** + ```bash + streamlit run ai_system_architect_r1.py + ``` + +4. **Use the Interface** + - Enter API credentials in sidebar + - Structure your prompt with: + - Project Context + - Requirements + - Constraints + - Scale + - Security/Compliance needs + - View detailed analysis results + +## Example Test Prompts: + +### 1. Financial Trading Platform +"We need to build a high-frequency trading platform that processes market data streams, executes trades with sub-millisecond latency, maintains audit trails, and handles complex risk calculations. The system needs to be globally distributed, handle 100,000 transactions per second, and have robust disaster recovery capabilities." +### 2. Multi-tenant SaaS Platform +"Design a multi-tenant SaaS platform for enterprise resource planning that needs to support customization per tenant, handle different data residency requirements, support offline capabilities, and maintain performance isolation between tenants. The system should scale to 10,000 concurrent users and support custom integrations." + +## Notes + +- Requires both DeepSeek and Anthropic API keys +- Provides real-time analysis with detailed explanations +- Supports chat-based interaction +- Includes clear reasoning for all architectural decisions +- API usage costs apply + + diff --git a/ai_agent_tutorials/ai_system_architect_r1/ai_system_architect_r1.py b/ai_agent_tutorials/ai_system_architect_r1/ai_system_architect_r1.py new file mode 100644 index 0000000..54efe12 --- /dev/null +++ b/ai_agent_tutorials/ai_system_architect_r1/ai_system_architect_r1.py @@ -0,0 +1,311 @@ +from typing import Optional, List, Dict, Any, Union +import os +import time +import streamlit as st +from openai import OpenAI +import anthropic +from dotenv import load_dotenv +from pydantic import BaseModel, Field +from enum import Enum +import json +from phi.agent import Agent, RunResponse +from phi.model.anthropic import Claude + +# Model Constants +DEEPSEEK_MODEL: str = "deepseek-reasoner" +CLAUDE_MODEL: str = "claude-3-5-sonnet-20241022" + +# Load environment variables +load_dotenv() + + +class ArchitecturePattern(str, Enum): + """Architectural patterns for system design.""" + MICROSERVICES = "microservices" # Decomposed into small, independent services + MONOLITHIC = "monolithic" # Single, unified codebase + SERVERLESS = "serverless" # Function-as-a-Service architecture + EVENT_DRIVEN = "event_driven" # Asynchronous event-based communication + +class DatabaseType(str, Enum): + """Types of database systems.""" + SQL = "sql" # Relational databases with ACID properties + NOSQL = "nosql" # Non-relational databases for flexible schemas + HYBRID = "hybrid" # Combined SQL and NoSQL approach + +class ComplianceStandard(str, Enum): + """Regulatory compliance standards.""" + HIPAA = "hipaa" # Healthcare data protection + GDPR = "gdpr" # EU data privacy regulation + SOC2 = "soc2" # Service organization security controls + ISO27001 = "iso27001" # Information security management + +class ArchitectureDecision(BaseModel): + """Represents architectural decisions and their justifications.""" + pattern: ArchitecturePattern + rationale: str = Field(..., min_length=50) # Detailed explanation for the choice + trade_offs: Dict[str, List[str]] = Field(..., alias="trade_offs") # Pros and cons + estimated_cost: Dict[str, float] # Cost breakdown + +class SecurityMeasure(BaseModel): + """Security controls and implementation details.""" + measure_type: str # Type of security measure + implementation_priority: int = Field(..., ge=1, le=5) # Priority level 1-5 + compliance_standards: List[ComplianceStandard] # Applicable standards + data_classification: str # Data sensitivity level + +class InfrastructureResource(BaseModel): + """Infrastructure components and specifications.""" + resource_type: str # Type of infrastructure resource + specifications: Dict[str, str] # Technical specifications + scaling_policy: Dict[str, str] # Scaling rules and thresholds + estimated_cost: float # Estimated cost per resource + +class TechnicalAnalysis(BaseModel): + """Complete technical analysis of the system architecture.""" + architecture_decision: ArchitectureDecision # Core architecture choices + infrastructure_resources: List[InfrastructureResource] # Required resources + security_measures: List[SecurityMeasure] # Security controls + database_choice: DatabaseType # Database architecture + compliance_requirements: List[ComplianceStandard] = [] # Required standards + performance_requirements: List[Dict[str, Union[str, float]]] = [] # Performance metrics + risk_assessment: Dict[str, str] = {} # Identified risks and mitigations + + +class ModelChain: + def __init__(self, deepseek_api_key: str, anthropic_api_key: str) -> None: + self.client = OpenAI( + api_key=deepseek_api_key, + base_url="https://api.deepseek.com" + ) + self.claude_client = anthropic.Anthropic(api_key=anthropic_api_key) + self.agent = Agent( + model=Claude(id="claude-3-5-sonnet-20241022", api_key=anthropic_api_key), + system_prompt="""Given the user's query and the DeepSeek reasoning: + 1. Provide a detailed analysis of the architecture decisions + 2. Generate a project implementation roadmap + 3. Create a comprehensive technical specification document + 4. Format the output in clean markdown with proper sections + 5. Include diagrams descriptions in mermaid.js format""", + markdown=True + ) + + self.deepseek_messages: List[Dict[str, str]] = [] + self.claude_messages: List[Dict[str, Any]] = [] + self.current_model: str = CLAUDE_MODEL + def get_deepseek_reasoning(self, user_input: str) -> tuple[str, str]: + start_time = time.time() + + system_prompt = """You are an expert software architect and technical advisor. Analyze the user's project requirements + and provide structured reasoning about architecture, tools, and implementation strategies. + + IMPORTANT: Reason why you are choosing a particular architecture pattern, database type, etc. for user understanding in your reasoning. + + IMPORTANT: Your response must be a valid JSON object (not a string or any other format) that matches the schema provided below. + Do not include any explanatory text, markdown formatting, or code blocks - only return the JSON object. + + Schema: + { + "architecture_decision": { + "pattern": "one of: microservices|monolithic|serverless|event_driven|layered", + "rationale": "string", + "trade_offs": {"advantage": ["list of strings"], "disadvantage": ["list of strings"]}, + "estimated_cost": {"implementation": float, "maintenance": float} + }, + "infrastructure_resources": [{ + "resource_type": "string", + "specifications": {"key": "value"}, + "scaling_policy": {"key": "value"}, + "estimated_cost": float + }], + "security_measures": [{ + "measure_type": "string", + "implementation_priority": "integer 1-5", + "compliance_standards": ["hipaa", "gdpr", "soc2", "hitech", "iso27001", "pci_dss"], + "estimated_setup_time_days": "integer", + "data_classification": "one of: protected_health_information|personally_identifiable_information|confidential|public", + "encryption_requirements": {"key": "value"}, + "access_control_policy": {"role": ["permissions"]}, + "audit_requirements": ["list of strings"] + }], + "database_choice": "one of: sql|nosql|graph|time_series|hybrid", + "ml_capabilities": [{ + "model_type": "string", + "training_frequency": "string", + "input_data_types": ["list of strings"], + "performance_requirements": {"metric": float}, + "hardware_requirements": {"resource": "specification"}, + "regulatory_constraints": ["list of strings"] + }], + "data_integrations": [{ + "integration_type": "one of: hl7|fhir|dicom|rest|soap|custom", + "data_format": "string", + "frequency": "string", + "volume": "string", + "security_requirements": {"key": "value"} + }], + "performance_requirements": [{ + "metric_name": "string", + "target_value": float, + "measurement_unit": "string", + "priority": "integer 1-5" + }], + "audit_config": { + "log_retention_period": "integer", + "audit_events": ["list of strings"], + "compliance_mapping": {"standard": ["requirements"]} + }, + "api_config": { + "version": "string", + "auth_method": "string", + "rate_limits": {"role": "requests_per_minute"}, + "documentation_url": "string" + }, + "error_handling": { + "retry_policy": {"key": "value"}, + "fallback_strategies": ["list of strings"], + "notification_channels": ["list of strings"] + }, + "estimated_team_size": "integer", + "critical_path_components": ["list of strings"], + "risk_assessment": {"risk": "mitigation"}, + "maintenance_considerations": ["list of strings"], + "compliance_requirements": ["list of compliance standards"], + "data_retention_policy": {"data_type": "retention_period"}, + "disaster_recovery": {"key": "value"}, + "interoperability_standards": ["list of strings"] + } + + Consider scalability, security, maintenance, and technical debt in your analysis. + Focus on practical, modern solutions while being mindful of trade-offs.""" + + try: + deepseek_response = self.client.chat.completions.create( + model="deepseek-reasoner", + messages=[ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": user_input} + ], + max_tokens=3000, + stream=False + ) + + reasoning_content = deepseek_response.choices[0].message.reasoning_content + normal_content = deepseek_response.choices[0].message.content + + # Display the reasoning separately + with st.expander("DeepSeek Reasoning", expanded=True): + st.markdown(reasoning_content) + + + with st.expander("💭 Technical Analysis", expanded=True): + st.markdown(normal_content) + elapsed_time = time.time() - start_time + time_str = f"{elapsed_time/60:.1f} minutes" if elapsed_time >= 60 else f"{elapsed_time:.1f} seconds" + st.caption(f"⏱️ Analysis completed in {time_str}") + + # Return both reasoning and normal content + return reasoning_content, normal_content + + except Exception as e: + st.error(f"Error in DeepSeek analysis: {str(e)}") + return "Error occurred while analyzing", "" + + def get_claude_response(self, user_input: str, deepseek_output: tuple[str, str]) -> str: + try: + reasoning_content, normal_content = deepseek_output + + # Create expander for Claude's response + with st.expander("🤖 Claude's Response", expanded=True): + response_placeholder = st.empty() + + # Prepare the message with user input, reasoning and normal output + message = f"""User Query: {user_input} + + DeepSeek Reasoning: {reasoning_content} + + DeepSeek Technical Analysis: {normal_content} + Give detailed explanation for each key value pair in brief in the JSON object, and why we chose it clearly. Dont use your own opinions, use the reasoning and the structured output to explain the choices.""" + + # Use Phi Agent to get response + response: RunResponse = self.agent.run( + message=message + ) + + dub = response.content + st.markdown(dub) + return dub + + except Exception as e: + st.error(f"Error in Claude response: {str(e)}") + return "Error occurred while getting response" + +def main() -> None: + """Main function to run the Streamlit app.""" + st.title("🤖 AI System Architect Advisor with R1") + + # Add prompt guidance + st.info(""" + 📝 For best results, structure your prompt with: + + 1. **Project Context**: Brief description of your project/system + 2. **Requirements**: Key functional and non-functional requirements + 3. **Constraints**: Any technical, budget, or time constraints + 4. **Scale**: Expected user base and growth projections + 5. **Security/Compliance**: Any specific security or regulatory needs + + Example: + ``` + I need to build a healthcare data management system that: + - Handles patient records and appointments + - Needs to scale to 10,000 users + - Must be HIPAA compliant + - Budget constraint of $50k for initial setup + - Should integrate with existing hospital systems + ``` + """) + + # Sidebar for API keys + with st.sidebar: + st.header("⚙️ Configuration") + deepseek_api_key = st.text_input("DeepSeek API Key", type="password") + anthropic_api_key = st.text_input("Anthropic API Key", type="password") + + if st.button("🗑️ Clear Chat History"): + st.session_state.messages = [] + st.rerun() + + # Initialize session state for messages + if "messages" not in st.session_state: + st.session_state.messages = [] + + # Display chat messages + for message in st.session_state.messages: + with st.chat_message(message["role"]): + st.markdown(message["content"]) + + # Chat input + if prompt := st.chat_input("What would you like to know?"): + if not deepseek_api_key or not anthropic_api_key: + st.error("⚠️ Please enter both API keys in the sidebar.") + return + + # Initialize ModelChain + chain = ModelChain(deepseek_api_key, anthropic_api_key) + + # Add user message to chat + st.session_state.messages.append({"role": "user", "content": prompt}) + with st.chat_message("user"): + st.markdown(prompt) + + # Get AI response + with st.chat_message("assistant"): + with st.spinner("🤔 Thinking..."): + deepseek_output = chain.get_deepseek_reasoning(prompt) + + + with st.spinner("✍️ Responding..."): + response = chain.get_claude_response(prompt, deepseek_output) + st.session_state.messages.append({"role": "assistant", "content": response}) + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/ai_agent_tutorials/ai_system_architect_r1/requirements.txt b/ai_agent_tutorials/ai_system_architect_r1/requirements.txt new file mode 100644 index 0000000..63db0b3 --- /dev/null +++ b/ai_agent_tutorials/ai_system_architect_r1/requirements.txt @@ -0,0 +1,4 @@ +streamlit +openai +anthropic +phidata From 5bd522728862435c797b31fe5586efccf6498f9d Mon Sep 17 00:00:00 2001 From: Madhu Date: Sun, 2 Feb 2025 01:17:12 +0530 Subject: [PATCH 12/13] deletion of unnecessary file --- .../ai_r1-tooluse-langroid/README.md | 35 --- .../ai_r1-tooluse-langroid/main.py | 295 ------------------ .../ai_r1-tooluse-langroid/requirements.txt | 4 - .../ai_r1-tooluse-langroid/test.py | 184 ----------- .../ai_system_architect_r1.py | 4 - 5 files changed, 522 deletions(-) delete mode 100644 ai_agent_tutorials/ai_r1-tooluse-langroid/README.md delete mode 100644 ai_agent_tutorials/ai_r1-tooluse-langroid/main.py delete mode 100644 ai_agent_tutorials/ai_r1-tooluse-langroid/requirements.txt delete mode 100644 ai_agent_tutorials/ai_r1-tooluse-langroid/test.py diff --git a/ai_agent_tutorials/ai_r1-tooluse-langroid/README.md b/ai_agent_tutorials/ai_r1-tooluse-langroid/README.md deleted file mode 100644 index 6d87421..0000000 --- a/ai_agent_tutorials/ai_r1-tooluse-langroid/README.md +++ /dev/null @@ -1,35 +0,0 @@ -## Example Test Prompts - -### 1. Real-time Event Processing System -"We're building a real-time event processing system for a smart city infrastructure that needs to handle 100,000 IoT sensors, process environmental data, manage traffic flows, and provide emergency response capabilities. The system needs to be highly available, handle peak loads during emergencies, and maintain data integrity. Budget constraints exist but reliability is critical." - -### 2. Healthcare Data Platform -"Design a HIPAA-compliant healthcare data platform that needs to integrate with legacy systems, handle real-time patient monitoring, support ML-based diagnostics, and manage secure data sharing between different healthcare providers. The system should scale to handle data from 50 hospitals and support both real-time analytics and batch processing." - -### 3. Financial Trading Platform -"We need to build a high-frequency trading platform that processes market data streams, executes trades with sub-millisecond latency, maintains audit trails, and handles complex risk calculations. The system needs to be globally distributed, handle 100,000 transactions per second, and have robust disaster recovery capabilities." - -### 4. Multi-tenant SaaS Platform -"Design a multi-tenant SaaS platform for enterprise resource planning that needs to support customization per tenant, handle different data residency requirements, support offline capabilities, and maintain performance isolation between tenants. The system should scale to 10,000 concurrent users and support custom integrations." - -### 5. Digital Content Delivery Network -"We're building a global content delivery platform for streaming high-definition video content, supporting live streaming, VOD, and interactive content. The system needs to handle dynamic transcoding, support DRM, manage user-generated content, and optimize delivery based on network conditions and device capabilities." - -### 6. Supply Chain Management System -"Design a blockchain-based supply chain management system that needs to track products from source to retail, integrate with IoT sensors for condition monitoring, support smart contracts for automated settlements, and provide real-time visibility across the supply chain. The system should handle 1000 partners and support regulatory compliance reporting." - -Each of these prompts presents complex architectural challenges that require careful consideration of: -- Scalability patterns -- Data consistency requirements -- Security and compliance needs -- Integration complexities -- Performance optimization -- Cost-benefit trade-offs -- Technical debt implications -- Team expertise requirements - -The DeepSeek model will analyze these requirements and provide structured recommendations using the ProjectAnalysis schema, which Claude can then use to provide detailed implementation guidance. -Startups making critical technical decisions -Enterprise architecture modernization projects - - diff --git a/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py b/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py deleted file mode 100644 index 9be0460..0000000 --- a/ai_agent_tutorials/ai_r1-tooluse-langroid/main.py +++ /dev/null @@ -1,295 +0,0 @@ -from typing import Optional, List, Dict, Any, Union -import os -import time -import streamlit as st -from openai import OpenAI -import anthropic -from dotenv import load_dotenv -from pydantic import BaseModel, Field -from enum import Enum -import json -from phi.agent import Agent, RunResponse -from phi.model.anthropic import Claude - -# Model Constants -DEEPSEEK_MODEL: str = "deepseek-reasoner" -CLAUDE_MODEL: str = "claude-3-5-sonnet-20241022" - -# Load environment variables -load_dotenv() - -system_prompt = """You are a Senior Software Expert and Technical Documentation Assistant. Your role is to analyze the structured JSON response from DeepSeek, which contains architectural and technical recommendations across various domains, along with the original user query describing the software system they want to build. - - The input consists of: - - The user's original query describing their software requirements - - A structured JSON response containing recommendations for architecture, security, infrastructure, compliance and other technical domains - - For each key-value pair in the JSON: - 1. Present the key and its corresponding value in a readable report format - 2. Format the information in a clear, organized way - 3. Do not add your own opinions or suggestions - 4. Do not modify or reinterpret the provided information - - Keep your responses factual and directly based on the JSON content provided.""" - -class ArchitecturePattern(str, Enum): - """Architectural patterns for system design.""" - MICROSERVICES = "microservices" # Decomposed into small, independent services - MONOLITHIC = "monolithic" # Single, unified codebase - SERVERLESS = "serverless" # Function-as-a-Service architecture - EVENT_DRIVEN = "event_driven" # Asynchronous event-based communication - -class DatabaseType(str, Enum): - """Types of database systems.""" - SQL = "sql" # Relational databases with ACID properties - NOSQL = "nosql" # Non-relational databases for flexible schemas - HYBRID = "hybrid" # Combined SQL and NoSQL approach - -class ComplianceStandard(str, Enum): - """Regulatory compliance standards.""" - HIPAA = "hipaa" # Healthcare data protection - GDPR = "gdpr" # EU data privacy regulation - SOC2 = "soc2" # Service organization security controls - ISO27001 = "iso27001" # Information security management - -class ArchitectureDecision(BaseModel): - """Represents architectural decisions and their justifications.""" - pattern: ArchitecturePattern - rationale: str = Field(..., min_length=50) # Detailed explanation for the choice - trade_offs: Dict[str, List[str]] = Field(..., alias="trade_offs") # Pros and cons - estimated_cost: Dict[str, float] # Cost breakdown - -class SecurityMeasure(BaseModel): - """Security controls and implementation details.""" - measure_type: str # Type of security measure - implementation_priority: int = Field(..., ge=1, le=5) # Priority level 1-5 - compliance_standards: List[ComplianceStandard] # Applicable standards - data_classification: str # Data sensitivity level - -class InfrastructureResource(BaseModel): - """Infrastructure components and specifications.""" - resource_type: str # Type of infrastructure resource - specifications: Dict[str, str] # Technical specifications - scaling_policy: Dict[str, str] # Scaling rules and thresholds - estimated_cost: float # Estimated cost per resource - -class TechnicalAnalysis(BaseModel): - """Complete technical analysis of the system architecture.""" - architecture_decision: ArchitectureDecision # Core architecture choices - infrastructure_resources: List[InfrastructureResource] # Required resources - security_measures: List[SecurityMeasure] # Security controls - database_choice: DatabaseType # Database architecture - compliance_requirements: List[ComplianceStandard] = [] # Required standards - performance_requirements: List[Dict[str, Union[str, float]]] = [] # Performance metrics - risk_assessment: Dict[str, str] = {} # Identified risks and mitigations - - -class ModelChain: - def __init__(self, deepseek_api_key: str, anthropic_api_key: str) -> None: - self.client = OpenAI( - api_key=deepseek_api_key, - base_url="https://api.deepseek.com" - ) - self.claude_client = anthropic.Anthropic(api_key=anthropic_api_key) - self.agent = Agent( - model=Claude(id="claude-3-5-sonnet-20241022", api_key=anthropic_api_key), - system_prompt=system_prompt, - markdown=True - ) - - self.deepseek_messages: List[Dict[str, str]] = [] - self.claude_messages: List[Dict[str, Any]] = [] - self.current_model: str = CLAUDE_MODEL - def get_deepseek_reasoning(self, user_input: str) -> tuple[str, str]: - start_time = time.time() - - system_prompt = """You are an expert software architect and technical advisor. Analyze the user's project requirements - and provide structured reasoning about architecture, tools, and implementation strategies. - - IMPORTANT: Reason why you are choosing a particular architecture pattern, database type, etc. for user understanding in your reasoning. - - IMPORTANT: Your response must be a valid JSON object (not a string or any other format) that matches the schema provided below. - Do not include any explanatory text, markdown formatting, or code blocks - only return the JSON object. - - Schema: - { - "architecture_decision": { - "pattern": "one of: microservices|monolithic|serverless|event_driven|layered", - "rationale": "string", - "trade_offs": {"advantage": ["list of strings"], "disadvantage": ["list of strings"]}, - "estimated_cost": {"implementation": float, "maintenance": float} - }, - "infrastructure_resources": [{ - "resource_type": "string", - "specifications": {"key": "value"}, - "scaling_policy": {"key": "value"}, - "estimated_cost": float - }], - "security_measures": [{ - "measure_type": "string", - "implementation_priority": "integer 1-5", - "compliance_standards": ["hipaa", "gdpr", "soc2", "hitech", "iso27001", "pci_dss"], - "estimated_setup_time_days": "integer", - "data_classification": "one of: protected_health_information|personally_identifiable_information|confidential|public", - "encryption_requirements": {"key": "value"}, - "access_control_policy": {"role": ["permissions"]}, - "audit_requirements": ["list of strings"] - }], - "database_choice": "one of: sql|nosql|graph|time_series|hybrid", - "ml_capabilities": [{ - "model_type": "string", - "training_frequency": "string", - "input_data_types": ["list of strings"], - "performance_requirements": {"metric": float}, - "hardware_requirements": {"resource": "specification"}, - "regulatory_constraints": ["list of strings"] - }], - "data_integrations": [{ - "integration_type": "one of: hl7|fhir|dicom|rest|soap|custom", - "data_format": "string", - "frequency": "string", - "volume": "string", - "security_requirements": {"key": "value"} - }], - "performance_requirements": [{ - "metric_name": "string", - "target_value": float, - "measurement_unit": "string", - "priority": "integer 1-5" - }], - "audit_config": { - "log_retention_period": "integer", - "audit_events": ["list of strings"], - "compliance_mapping": {"standard": ["requirements"]} - }, - "api_config": { - "version": "string", - "auth_method": "string", - "rate_limits": {"role": "requests_per_minute"}, - "documentation_url": "string" - }, - "error_handling": { - "retry_policy": {"key": "value"}, - "fallback_strategies": ["list of strings"], - "notification_channels": ["list of strings"] - }, - "estimated_team_size": "integer", - "critical_path_components": ["list of strings"], - "risk_assessment": {"risk": "mitigation"}, - "maintenance_considerations": ["list of strings"], - "compliance_requirements": ["list of compliance standards"], - "data_retention_policy": {"data_type": "retention_period"}, - "disaster_recovery": {"key": "value"}, - "interoperability_standards": ["list of strings"] - } - - Consider scalability, security, maintenance, and technical debt in your analysis. - Focus on practical, modern solutions while being mindful of trade-offs.""" - - try: - deepseek_response = self.client.chat.completions.create( - model="deepseek-reasoner", - messages=[ - {"role": "system", "content": system_prompt}, - {"role": "user", "content": user_input} - ], - max_tokens=3000, - stream=False - ) - - reasoning_content = deepseek_response.choices[0].message.reasoning_content - normal_content = deepseek_response.choices[0].message.content - - # Display the reasoning separately - with st.expander("DeepSeek Reasoning", expanded=True): - st.markdown(reasoning_content) - - - with st.expander("💭 Technical Analysis", expanded=True): - st.markdown(normal_content) - elapsed_time = time.time() - start_time - time_str = f"{elapsed_time/60:.1f} minutes" if elapsed_time >= 60 else f"{elapsed_time:.1f} seconds" - st.caption(f"⏱️ Analysis completed in {time_str}") - - # Return both reasoning and normal content - return reasoning_content, normal_content - - except Exception as e: - st.error(f"Error in DeepSeek analysis: {str(e)}") - return "Error occurred while analyzing", "" - - def get_claude_response(self, user_input: str, deepseek_output: tuple[str, str]) -> str: - try: - reasoning_content, normal_content = deepseek_output - - # Create expander for Claude's response - with st.expander("🤖 Claude's Response", expanded=True): - response_placeholder = st.empty() - - # Prepare the message with user input, reasoning and normal output - message = f"""User Query: {user_input} - - DeepSeek Reasoning: {reasoning_content} - - DeepSeek Technical Analysis: {normal_content}""" - - # Use Phi Agent to get response - response: RunResponse = self.agent.run( - message=message - ) - - return response.content - - except Exception as e: - st.error(f"Error in Claude response: {str(e)}") - return "Error occurred while getting response" - -def main() -> None: - """Main function to run the Streamlit app.""" - st.title("🤖 AI Project with Deepseek + R1") - - # Sidebar for API keys - with st.sidebar: - st.header("⚙️ Configuration") - deepseek_api_key = st.text_input("DeepSeek API Key", type="password") - anthropic_api_key = st.text_input("Anthropic API Key", type="password") - - if st.button("🗑️ Clear Chat History"): - st.session_state.messages = [] - st.rerun() - - # Initialize session state for messages - if "messages" not in st.session_state: - st.session_state.messages = [] - - # Display chat messages - for message in st.session_state.messages: - with st.chat_message(message["role"]): - st.markdown(message["content"]) - - # Chat input - if prompt := st.chat_input("What would you like to know?"): - if not deepseek_api_key or not anthropic_api_key: - st.error("⚠️ Please enter both API keys in the sidebar.") - return - - # Initialize ModelChain - chain = ModelChain(deepseek_api_key, anthropic_api_key) - - # Add user message to chat - st.session_state.messages.append({"role": "user", "content": prompt}) - with st.chat_message("user"): - st.markdown(prompt) - - # Get AI response - with st.chat_message("assistant"): - with st.spinner("🤔 Thinking..."): - deepseek_output = chain.get_deepseek_reasoning(prompt) - - - with st.spinner("✍️ Responding..."): - response = chain.get_claude_response(prompt, deepseek_output) - st.session_state.messages.append({"role": "assistant", "content": response}) - -if __name__ == "__main__": - main() \ No newline at end of file diff --git a/ai_agent_tutorials/ai_r1-tooluse-langroid/requirements.txt b/ai_agent_tutorials/ai_r1-tooluse-langroid/requirements.txt deleted file mode 100644 index 5dc79a4..0000000 --- a/ai_agent_tutorials/ai_r1-tooluse-langroid/requirements.txt +++ /dev/null @@ -1,4 +0,0 @@ -streamlit -openai -anthropic -python-dotenv diff --git a/ai_agent_tutorials/ai_r1-tooluse-langroid/test.py b/ai_agent_tutorials/ai_r1-tooluse-langroid/test.py deleted file mode 100644 index 35a9c40..0000000 --- a/ai_agent_tutorials/ai_r1-tooluse-langroid/test.py +++ /dev/null @@ -1,184 +0,0 @@ -from enum import Enum -from typing import List, Dict, Union -from pydantic import BaseModel, Field, ValidationError -import streamlit as st -from openai import OpenAI -import anthropic -import json -import re -import os -from dotenv import load_dotenv -from phi.agent import Agent, RunResponse -from phi.model.anthropic import Claude - -load_dotenv() - -# -------------------------- -# Enums & Data Models -# -------------------------- -class ArchitecturePattern(str, Enum): - MICROSERVICES = "microservices" - MONOLITHIC = "monolithic" - SERVERLESS = "serverless" - EVENT_DRIVEN = "event_driven" - -class DatabaseType(str, Enum): - SQL = "sql" - NOSQL = "nosql" - HYBRID = "hybrid" - -class ComplianceStandard(str, Enum): - HIPAA = "hipaa" - GDPR = "gdpr" - SOC2 = "soc2" - ISO27001 = "iso27001" - -class ArchitectureDecision(BaseModel): - pattern: ArchitecturePattern - rationale: str = Field(..., min_length=50) - trade_offs: Dict[str, List[str]] = Field(..., alias="trade_offs") - estimated_cost: Dict[str, float] - -class SecurityMeasure(BaseModel): - measure_type: str - implementation_priority: int = Field(..., ge=1, le=5) - compliance_standards: List[ComplianceStandard] - data_classification: str - -class InfrastructureResource(BaseModel): - resource_type: str - specifications: Dict[str, str] - scaling_policy: Dict[str, str] - estimated_cost: float - -class TechnicalAnalysis(BaseModel): - architecture_decision: ArchitectureDecision - infrastructure_resources: List[InfrastructureResource] - security_measures: List[SecurityMeasure] - database_choice: DatabaseType - compliance_requirements: List[ComplianceStandard] = [] - performance_requirements: List[Dict[str, Union[str, float]]] = [] - risk_assessment: Dict[str, str] = {} - -# -------------------------- -# Core Implementation -# -------------------------- -class ArchitectureAnalyzer: - def __init__(self, deepseek_api_key: str, anthropic_api_key: str): - self.deepseek_client = OpenAI( - api_key=deepseek_api_key, - base_url="https://api.deepseek.com" - ) - self.claude_agent = Agent( - model=Claude( - id="claude-3-5-sonnet-20241022", - api_key=anthropic_api_key - ), - markdown=True, - ) - self.reasoning_content = "" - - self.deepseek_prompt = f"""Analyze software requirements and return JSON with: -{{ - "architecture_decision": {{ - "pattern": "{'|'.join([e.value for e in ArchitecturePattern])}", - "rationale": "technical justification", - "trade_offs": {{"pros": [], "cons": []}}, - "estimated_cost": {{"development": float, "maintenance": float}} - }}, - "infrastructure_resources": [{{"resource_type": "...", "specifications": {{}}, ...}}], - "security_measures": [{{"measure_type": "...", "priority": 1-5, ...}}], - "database_choice": "{'|'.join([e.value for e in DatabaseType])}", - "compliance_requirements": ["..."], - "performance_requirements": [{{"metric": "...", "target": float}}] -}}""" - - def _extract_json(self, text: str) -> dict: - try: - json_str = re.search(r'\{.*\}', text, re.DOTALL).group() - return json.loads(json_str) - except (AttributeError, json.JSONDecodeError) as e: - st.error(f"JSON extraction failed: {str(e)}") - st.text("Raw response:\n" + text) - raise - - def analyze_requirements(self, user_input: str) -> TechnicalAnalysis: - try: - response1 = self.deepseek_client.chat.completions.create( - model="deepseek-reasoner", - messages=[ - {"role": "system", "content": self.deepseek_prompt}, - {"role": "user", "content": user_input} - ], - temperature=0.2, - max_tokens=2000 - ) - self.reasoning_content = response1.choices[0].message.reasoning_content - json_data = self._extract_json(response1.choices[0].message.content) - return TechnicalAnalysis(**json_data) - - except ValidationError as e: - st.error(f"Validation error: {e.errors()}") - st.json(json_data) - raise - - def generate_report(self, analysis: TechnicalAnalysis) -> str: - report_prompt = f"""Convert this technical analysis into a executive report: -{analysis.model_dump_json(indent=2)} - -Use markdown with: -# Title -## Sections -- Bullet points -**Bold important items** -Tables for cost/performance""" - - response = self.claude_agent.run(report_prompt) - return response.content - -# -------------------------- -# Streamlit UI -# -------------------------- -def main(): - st.title("🏗️ AI Architecture Advisor") - - with st.sidebar: - st.header("🔑 Setup") - deepseek_api_key = st.text_input("DeepSeek Key", type="password") - anthropic_api_key = st.text_input("Claude Key", type="password") - - if "analysis" not in st.session_state: - st.session_state.analysis = None - - if prompt := st.chat_input("Describe your system requirements:"): - if not all([deepseek_api_key, anthropic_api_key]): - st.error("Missing API keys") - return - - analyzer = ArchitectureAnalyzer(deepseek_api_key, anthropic_api_key) - - with st.status("🔨 Processing...", expanded=True): - try: - # Analysis Phase - st.write("🧠 Analyzing requirements...") - analysis = analyzer.analyze_requirements(prompt) - st.session_state.analysis = analysis - with st.expander("reasoning"): - st.markdown(analyzer.reasoning_content) - - # Reporting Phase - st.write("📊 Generating report...") - report = analyzer.generate_report(analysis) - - # Display Results - st.success("Analysis complete!") - st.markdown(report) - - with st.expander("📁 Raw Analysis Data"): - st.json(analysis.model_dump_json()) - - except Exception as e: - st.error(f"Processing failed: {str(e)}") - -if __name__ == "__main__": - main() \ No newline at end of file diff --git a/ai_agent_tutorials/ai_system_architect_r1/ai_system_architect_r1.py b/ai_agent_tutorials/ai_system_architect_r1/ai_system_architect_r1.py index 54efe12..fb81877 100644 --- a/ai_agent_tutorials/ai_system_architect_r1/ai_system_architect_r1.py +++ b/ai_agent_tutorials/ai_system_architect_r1/ai_system_architect_r1.py @@ -15,10 +15,6 @@ from phi.model.anthropic import Claude DEEPSEEK_MODEL: str = "deepseek-reasoner" CLAUDE_MODEL: str = "claude-3-5-sonnet-20241022" -# Load environment variables -load_dotenv() - - class ArchitecturePattern(str, Enum): """Architectural patterns for system design.""" MICROSERVICES = "microservices" # Decomposed into small, independent services From df784d9c00fffb4eab834c46bf393e87682fbc53 Mon Sep 17 00:00:00 2001 From: Madhu Date: Sun, 2 Feb 2025 19:32:29 +0530 Subject: [PATCH 13/13] phidata -> agno --- .../ai_system_architect_r1/README.md | 2 +- .../ai_system_architect_r1.py | 18 +++++++++++++----- .../ai_system_architect_r1/requirements.txt | 2 +- 3 files changed, 15 insertions(+), 7 deletions(-) diff --git a/ai_agent_tutorials/ai_system_architect_r1/README.md b/ai_agent_tutorials/ai_system_architect_r1/README.md index 4c3a314..a78e925 100644 --- a/ai_agent_tutorials/ai_system_architect_r1/README.md +++ b/ai_agent_tutorials/ai_system_architect_r1/README.md @@ -1,6 +1,6 @@ # 🤖 AI System Architect Advisor with R1 -A Streamlit application that provides expert software architecture analysis and recommendations using a dual-model approach combining DeepSeek R1's Reasoning and Claude. The system provides detailed technical analysis, implementation roadmaps, and architectural decisions for complex software systems. +An Agno agentic system that provides expert software architecture analysis and recommendations using a dual-model approach combining DeepSeek R1's Reasoning and Claude. The system provides detailed technical analysis, implementation roadmaps, and architectural decisions for complex software systems. ## Features diff --git a/ai_agent_tutorials/ai_system_architect_r1/ai_system_architect_r1.py b/ai_agent_tutorials/ai_system_architect_r1/ai_system_architect_r1.py index fb81877..5236c94 100644 --- a/ai_agent_tutorials/ai_system_architect_r1/ai_system_architect_r1.py +++ b/ai_agent_tutorials/ai_system_architect_r1/ai_system_architect_r1.py @@ -8,8 +8,8 @@ from dotenv import load_dotenv from pydantic import BaseModel, Field from enum import Enum import json -from phi.agent import Agent, RunResponse -from phi.model.anthropic import Claude +from agno.agent import Agent, RunResponse +from agno.models.anthropic import Claude # Model Constants DEEPSEEK_MODEL: str = "deepseek-reasoner" @@ -74,14 +74,22 @@ class ModelChain: base_url="https://api.deepseek.com" ) self.claude_client = anthropic.Anthropic(api_key=anthropic_api_key) - self.agent = Agent( - model=Claude(id="claude-3-5-sonnet-20241022", api_key=anthropic_api_key), + + # Create Claude model with system prompt + claude_model = Claude( + id="claude-3-5-sonnet-20241022", + api_key=anthropic_api_key, system_prompt="""Given the user's query and the DeepSeek reasoning: 1. Provide a detailed analysis of the architecture decisions 2. Generate a project implementation roadmap 3. Create a comprehensive technical specification document 4. Format the output in clean markdown with proper sections - 5. Include diagrams descriptions in mermaid.js format""", + 5. Include diagrams descriptions in mermaid.js format""" + ) + + # Initialize agent with configured model + self.agent = Agent( + model=claude_model, markdown=True ) diff --git a/ai_agent_tutorials/ai_system_architect_r1/requirements.txt b/ai_agent_tutorials/ai_system_architect_r1/requirements.txt index 63db0b3..e8fce4b 100644 --- a/ai_agent_tutorials/ai_system_architect_r1/requirements.txt +++ b/ai_agent_tutorials/ai_system_architect_r1/requirements.txt @@ -1,4 +1,4 @@ streamlit openai anthropic -phidata +agno \ No newline at end of file