Introducing the LivinGrimoire Deepseek Async Skill

The LivinGrimoire framework (short for Living Grimoire) continues to push the boundaries of simplicity and power in building intelligent systems. Today, I’m excited to introduce a new async skill: the Deepseek Async Skill. This skill allows you to seamlessly integrate Deepseek’s capabilities into your LivinGrimoire system with just one line of code. Let’s dive into how it works and why it’s a game-changer.

What is the Deepseek Async Skill?

The Deepseek Async Skill is a new addition to LivinGrimoire that enables asynchronous communication with the Deepseek API. Since API calls can take a few seconds to complete, this skill runs in the background, ensuring your system remains responsive while waiting for the response.

The skill is designed to activate when the user’s input ends with the word "run". Once triggered, it sends the input (minus the word "run") to Deepseek and outputs the response when ready.

How It Works

Here’s the Python code that defines the DaDeepseekRun skill:

<span>class</span> <span>DaDeepseekRun</span><span>(</span><span>ShorniSplash</span><span>):</span>
    <span>def</span> <span>__init__</span><span>(</span><span>self</span><span>):</span>
        <span>super</span><span>().</span><span>__init__</span><span>()</span>
        <span>self</span><span>.</span><span>input_text</span> <span>=</span> <span>""</span>  <span># Temporary storage for input text </span>
    <span>def</span> <span>trigger</span><span>(</span><span>self</span><span>,</span> <span>ear</span><span>:</span> <span>str</span><span>,</span> <span>skin</span><span>:</span> <span>str</span><span>,</span> <span>eye</span><span>:</span> <span>str</span><span>)</span> <span>-></span> <span>bool</span><span>:</span>
        <span># Check if the ear string ends with the word "run" </span>        <span>return</span> <span>ear</span><span>.</span><span>strip</span><span>().</span><span>endswith</span><span>(</span><span>"</span><span>run</span><span>"</span><span>)</span>
    <span>@staticmethod</span>
    <span>def</span> <span>_async_func</span><span>(</span><span>this_cls</span><span>):</span>
        <span># Use the stored input text </span>        <span>input_text</span> <span>=</span> <span>this_cls</span><span>.</span><span>input_text</span>
        <span># Call the Deepseek API (replace with actual API endpoint and logic) </span>        <span>try</span><span>:</span>
            <span>response</span> <span>=</span> <span>this_cls</span><span>.</span><span>call_deepseek_api</span><span>(</span><span>input_text</span><span>)</span>
            <span>this_cls</span><span>.</span><span>_result</span> <span>=</span> <span>response</span>
        <span>except</span> <span>Exception</span> <span>as</span> <span>e</span><span>:</span>
            <span>this_cls</span><span>.</span><span>_result</span> <span>=</span> <span>f</span><span>"</span><span>Error calling Deepseek API: </span><span>{</span><span>str</span><span>(</span><span>e</span><span>)</span><span>}</span><span>"</span>
    <span>def</span> <span>input</span><span>(</span><span>self</span><span>,</span> <span>ear</span><span>:</span> <span>str</span><span>,</span> <span>skin</span><span>:</span> <span>str</span><span>,</span> <span>eye</span><span>:</span> <span>str</span><span>):</span>
        <span># Check if the skill should trigger </span>        <span>if</span> <span>self</span><span>.</span><span>trigger</span><span>(</span><span>ear</span><span>,</span> <span>skin</span><span>,</span> <span>eye</span><span>):</span>
            <span># Remove the last word "run" from the ear string </span>            <span>self</span><span>.</span><span>input_text</span> <span>=</span> <span>ear</span><span>.</span><span>rsplit</span><span>(</span><span>"</span><span> </span><span>"</span><span>,</span> <span>1</span><span>)[</span><span>0</span><span>].</span><span>strip</span><span>()</span>
            <span># Start the async operation in a daemon thread </span>            <span>my_thread</span> <span>=</span> <span>threading</span><span>.</span><span>Thread</span><span>(</span>
                <span>target</span><span>=</span><span>self</span><span>.</span><span>_async_func</span><span>,</span>
                <span>args</span><span>=</span><span>(</span><span>self</span><span>,)</span>  <span># Pass self as the only argument </span>            <span>)</span>
            <span>my_thread</span><span>.</span><span>daemon</span> <span>=</span> <span>True</span>
            <span>my_thread</span><span>.</span><span>start</span><span>()</span>
        <span># Output the result if available </span>        <span>if</span> <span>len</span><span>(</span><span>self</span><span>.</span><span>_result</span><span>)</span> <span>></span> <span>0</span><span>:</span>
            <span>self</span><span>.</span><span>output_result</span><span>()</span>
            <span>self</span><span>.</span><span>_result</span> <span>=</span> <span>""</span>
    <span>@staticmethod</span>
    <span>def</span> <span>call_deepseek_api</span><span>(</span><span>input_text</span><span>:</span> <span>str</span><span>)</span> <span>-></span> <span>str</span><span>:</span>
        <span># Replace this with the actual Deepseek API call logic </span>        <span># Example: </span>        <span># api_url = "https://api.deepseek.com/chat" </span>        <span># payload = {"input": input_text} </span>        <span># headers = {"Authorization": "Bearer YOUR_API_KEY"} </span>        <span># response = requests.post(api_url, json=payload, headers=headers) </span>        <span># return response.json().get("response", "No response from API") </span>
        <span># For now, just return a mock response </span>        <span>return</span> <span>f</span><span>"</span><span>Deepseek response to: </span><span>{</span><span>input_text</span><span>}</span><span>"</span>
<span>class</span> <span>DaDeepseekRun</span><span>(</span><span>ShorniSplash</span><span>):</span>
    <span>def</span> <span>__init__</span><span>(</span><span>self</span><span>):</span>
        <span>super</span><span>().</span><span>__init__</span><span>()</span>
        <span>self</span><span>.</span><span>input_text</span> <span>=</span> <span>""</span>  <span># Temporary storage for input text </span>
    <span>def</span> <span>trigger</span><span>(</span><span>self</span><span>,</span> <span>ear</span><span>:</span> <span>str</span><span>,</span> <span>skin</span><span>:</span> <span>str</span><span>,</span> <span>eye</span><span>:</span> <span>str</span><span>)</span> <span>-></span> <span>bool</span><span>:</span>
        <span># Check if the ear string ends with the word "run" </span>        <span>return</span> <span>ear</span><span>.</span><span>strip</span><span>().</span><span>endswith</span><span>(</span><span>"</span><span>run</span><span>"</span><span>)</span>

    <span>@staticmethod</span>
    <span>def</span> <span>_async_func</span><span>(</span><span>this_cls</span><span>):</span>
        <span># Use the stored input text </span>        <span>input_text</span> <span>=</span> <span>this_cls</span><span>.</span><span>input_text</span>

        <span># Call the Deepseek API (replace with actual API endpoint and logic) </span>        <span>try</span><span>:</span>
            <span>response</span> <span>=</span> <span>this_cls</span><span>.</span><span>call_deepseek_api</span><span>(</span><span>input_text</span><span>)</span>
            <span>this_cls</span><span>.</span><span>_result</span> <span>=</span> <span>response</span>
        <span>except</span> <span>Exception</span> <span>as</span> <span>e</span><span>:</span>
            <span>this_cls</span><span>.</span><span>_result</span> <span>=</span> <span>f</span><span>"</span><span>Error calling Deepseek API: </span><span>{</span><span>str</span><span>(</span><span>e</span><span>)</span><span>}</span><span>"</span>

    <span>def</span> <span>input</span><span>(</span><span>self</span><span>,</span> <span>ear</span><span>:</span> <span>str</span><span>,</span> <span>skin</span><span>:</span> <span>str</span><span>,</span> <span>eye</span><span>:</span> <span>str</span><span>):</span>
        <span># Check if the skill should trigger </span>        <span>if</span> <span>self</span><span>.</span><span>trigger</span><span>(</span><span>ear</span><span>,</span> <span>skin</span><span>,</span> <span>eye</span><span>):</span>
            <span># Remove the last word "run" from the ear string </span>            <span>self</span><span>.</span><span>input_text</span> <span>=</span> <span>ear</span><span>.</span><span>rsplit</span><span>(</span><span>"</span><span> </span><span>"</span><span>,</span> <span>1</span><span>)[</span><span>0</span><span>].</span><span>strip</span><span>()</span>

            <span># Start the async operation in a daemon thread </span>            <span>my_thread</span> <span>=</span> <span>threading</span><span>.</span><span>Thread</span><span>(</span>
                <span>target</span><span>=</span><span>self</span><span>.</span><span>_async_func</span><span>,</span>
                <span>args</span><span>=</span><span>(</span><span>self</span><span>,)</span>  <span># Pass self as the only argument </span>            <span>)</span>
            <span>my_thread</span><span>.</span><span>daemon</span> <span>=</span> <span>True</span>
            <span>my_thread</span><span>.</span><span>start</span><span>()</span>

        <span># Output the result if available </span>        <span>if</span> <span>len</span><span>(</span><span>self</span><span>.</span><span>_result</span><span>)</span> <span>></span> <span>0</span><span>:</span>
            <span>self</span><span>.</span><span>output_result</span><span>()</span>
            <span>self</span><span>.</span><span>_result</span> <span>=</span> <span>""</span>

    <span>@staticmethod</span>
    <span>def</span> <span>call_deepseek_api</span><span>(</span><span>input_text</span><span>:</span> <span>str</span><span>)</span> <span>-></span> <span>str</span><span>:</span>
        <span># Replace this with the actual Deepseek API call logic </span>        <span># Example: </span>        <span># api_url = "https://api.deepseek.com/chat" </span>        <span># payload = {"input": input_text} </span>        <span># headers = {"Authorization": "Bearer YOUR_API_KEY"} </span>        <span># response = requests.post(api_url, json=payload, headers=headers) </span>        <span># return response.json().get("response", "No response from API") </span>
        <span># For now, just return a mock response </span>        <span>return</span> <span>f</span><span>"</span><span>Deepseek response to: </span><span>{</span><span>input_text</span><span>}</span><span>"</span>
class DaDeepseekRun(ShorniSplash):
    def __init__(self):
        super().__init__()
        self.input_text = ""  # Temporary storage for input text 
    def trigger(self, ear: str, skin: str, eye: str) -> bool:
        # Check if the ear string ends with the word "run"         return ear.strip().endswith("run")

    @staticmethod
    def _async_func(this_cls):
        # Use the stored input text         input_text = this_cls.input_text

        # Call the Deepseek API (replace with actual API endpoint and logic)         try:
            response = this_cls.call_deepseek_api(input_text)
            this_cls._result = response
        except Exception as e:
            this_cls._result = f"Error calling Deepseek API: {str(e)}"

    def input(self, ear: str, skin: str, eye: str):
        # Check if the skill should trigger         if self.trigger(ear, skin, eye):
            # Remove the last word "run" from the ear string             self.input_text = ear.rsplit(" ", 1)[0].strip()

            # Start the async operation in a daemon thread             my_thread = threading.Thread(
                target=self._async_func,
                args=(self,)  # Pass self as the only argument             )
            my_thread.daemon = True
            my_thread.start()

        # Output the result if available         if len(self._result) > 0:
            self.output_result()
            self._result = ""

    @staticmethod
    def call_deepseek_api(input_text: str) -> str:
        # Replace this with the actual Deepseek API call logic         # Example:         # api_url = "https://api.deepseek.com/chat"         # payload = {"input": input_text}         # headers = {"Authorization": "Bearer YOUR_API_KEY"}         # response = requests.post(api_url, json=payload, headers=headers)         # return response.json().get("response", "No response from API") 
        # For now, just return a mock response         return f"Deepseek response to: {input_text}"

Enter fullscreen mode Exit fullscreen mode

Key Features

1. Trigger Condition:

   • The skill activates when the input string ends with the word "run". This is defined in the trigger method.

2. Asynchronous Execution:

   • When triggered, the skill removes the word "run" from the input and starts an asynchronous operation using a daemon thread. This ensures the main program remains responsive.

3. Deepseek API Integration:

   • The call_deepseek_api method is a placeholder for the actual API call. For now, it returns a mock response.

4. Result Handling:

   • Once the API call completes, the result is stored in _result and outputted using output_result().

Adding the Skill to LivinGrimoire

Here’s the best part: adding this skill to your LivinGrimoire system takes just one line of code:

<span>brain</span><span>.</span><span>add_logical_skill</span><span>(</span><span>DaDeepseekRun</span><span>())</span>
<span>brain</span><span>.</span><span>add_logical_skill</span><span>(</span><span>DaDeepseekRun</span><span>())</span>
brain.add_logical_skill(DaDeepseekRun())

Enter fullscreen mode Exit fullscreen mode

The Da prefix indicates that this is an asynchronous skill. Once added, the skill will automatically engage when the input ends with "run".

Why This is Awesome

1. Lazy Integration:

   • With just one line of code, you can absorb the power of Deepseek into your LivinGrimoire system. This is the epitome of lazy programming—maximizing functionality with minimal effort.

2. Asynchronous by Design:

   • Long-running tasks like API calls don’t block the main thread, keeping your system responsive.

3. Modular and Reusable:

   • Skills are self-contained and easy to add or remove. This modularity makes LivinGrimoire incredibly flexible.

Conclusion

The new Deepseek Async Skill is a testament to the power and simplicity of LivinGrimoire. By abstracting away the complexity of threading and asynchronous operations, it allows you to focus on building intelligent systems without getting bogged down in the details.

原文链接：deepseek with 1 line of code